Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 12 additions & 11 deletions kanboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,14 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import json
import asyncio
import base64
import functools
import asyncio
import json
import ssl
from typing import Dict, Optional
from typing import Any, Callable, Dict, Optional
from urllib import request as http


DEFAULT_AUTH_HEADER = "Authorization"
ASYNC_FUNCNAME_MARKER = "_async"

Expand Down Expand Up @@ -100,21 +99,23 @@ def __init__(
except RuntimeError:
self._event_loop = asyncio.new_event_loop()

def __getattr__(self, name: str):
def __getattr__(self, name: str) -> Callable[..., Any]:
if self.is_async_method_name(name):

async def function(*args, **kwargs):
async def function(*args: Any, **kwargs: Any) -> Any:
return await self._event_loop.run_in_executor(
None,
functools.partial(
self.execute, method=self._to_camel_case(self.get_funcname_from_async_name(name)), **kwargs
self.execute,
method=self._to_camel_case(self.get_funcname_from_async_name(name)),
**kwargs,
),
)

return function
else:

def function(*args, **kwargs):
def function(*args: Any, **kwargs: Any) -> Any:
return self.execute(method=self._to_camel_case(name), **kwargs)

return function
Expand All @@ -133,7 +134,7 @@ def _to_camel_case(snake_str: str) -> str:
return components[0] + "".join(x.title() for x in components[1:])

@staticmethod
def _parse_response(response: bytes):
def _parse_response(response: bytes) -> Any:
if not response:
raise ClientError("Empty response from server")
try:
Expand All @@ -147,7 +148,7 @@ def _parse_response(response: bytes):
except ValueError as e:
raise ClientError(f"Failed to parse JSON response: {e}")

def _do_request(self, headers: Dict[str, str], body: Dict):
def _do_request(self, headers: Dict[str, str], body: Dict[str, Any]) -> Any:
try:
request = http.Request(self._url, headers=headers, data=json.dumps(body).encode())

Expand All @@ -164,7 +165,7 @@ def _do_request(self, headers: Dict[str, str], body: Dict):
raise ClientError(str(e))
return self._parse_response(response)

def execute(self, method: str, **kwargs):
def execute(self, method: str, **kwargs: Any) -> Any:
"""
Call a remote Kanboard API procedure.

Expand Down