Skip to content

Commit

Permalink
Make mypy happy
Browse files Browse the repository at this point in the history
  • Loading branch information
FlorianWilhelm committed Dec 28, 2023
1 parent 4c81733 commit 990e0f0
Show file tree
Hide file tree
Showing 6 changed files with 68 additions and 47 deletions.
14 changes: 7 additions & 7 deletions src/pytanis/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,22 +13,22 @@
"""Path within $HOME to the configuration file of Pytanis"""


class Google(BaseModel):
class GoogleCfg(BaseModel):
"""Configuration related to the Google API"""

client_secret_json: Path | None = None
token_json: Path | None = None


class HelpDesk(BaseModel):
class HelpDeskCfg(BaseModel):
"""Configuration related to the HelpDesk API"""

account: str | None = None
entity_id: str | None = None
token: str | None = None


class Pretalx(BaseModel):
class PretalxCfg(BaseModel):
"""Configuration related to the Pretalx API"""

api_token: str | None = None
Expand All @@ -39,13 +39,13 @@ class Config(BaseModel):

cfg_path: FilePath

Pretalx: Pretalx
Google: Google
HelpDesk: HelpDesk
Pretalx: PretalxCfg
Google: GoogleCfg
HelpDesk: HelpDeskCfg

@field_validator('Google')
@classmethod
def convert_json_path(cls, v: Google, info: FieldValidationInfo) -> Google:
def convert_json_path(cls, v: GoogleCfg, info: FieldValidationInfo) -> GoogleCfg:
def make_rel_path_abs(entry):
if entry is not None and not entry.is_absolute():
entry = info.data['cfg_path'].parent / entry
Expand Down
30 changes: 24 additions & 6 deletions src/pytanis/helpdesk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,10 @@
* Transfer more functionality from https://github.com/PYCONDE/py_helpdesk_com
"""

from typing import Any, TypeAlias
from typing import Any, TypeAlias, cast

import httpx
from httpx import URL, Response
from httpx import URL, QueryParams, Response
from httpx_auth import Basic
from structlog import get_logger

Expand Down Expand Up @@ -47,27 +47,45 @@ def set_throttling(self, calls: int, seconds: int):
self._get_throttled = throttle(calls, seconds)(self._get)
self._post_throttled = throttle(calls, seconds)(self._post)

def _get(self, endpoint: str, params: dict[str, str] | None = None) -> Response:
def _get(self, endpoint: str, params: QueryParams | None = None) -> Response:
"""Retrieve data via raw GET request"""
if params is None:
params = cast(QueryParams, {})
if self._config.HelpDesk.token is None:
msg = 'API token for Helpdesk is empty'
raise RuntimeError(msg)
if self._config.HelpDesk.account is None:
msg = 'Account for Helpdesk is empty'
raise RuntimeError(msg)

auth = Basic(self._config.HelpDesk.account, self._config.HelpDesk.token)
url = URL('https://api.helpdesk.com/v1/').join(endpoint)
_logger.debug(f'GET: {url.copy_merge_params(params)}')
return httpx.get(url, auth=auth, params=params, headers=self._headers)

def get(self, endpoint: str, params: dict[str, str] | None = None) -> JSON:
def get(self, endpoint: str, params: QueryParams | None = None) -> JSON:
"""Retrieve data via throttled GET request and return the JSON"""
resp = self._get_throttled(endpoint, params)
resp.raise_for_status()
return resp.json()

def _post(self, endpoint: str, data: dict[str, Any], params: dict[str, str] | None = None) -> Response:
def _post(self, endpoint: str, data: dict[str, Any], params: QueryParams | None = None) -> Response:
"""Sent data via raw POST request"""
if params is None:
params = cast(QueryParams, {})
if self._config.HelpDesk.token is None:
msg = 'API token for Helpdesk is empty'
raise RuntimeError(msg)
if self._config.HelpDesk.account is None:
msg = 'Account for Helpdesk is empty'
raise RuntimeError(msg)

auth = Basic(self._config.HelpDesk.account, self._config.HelpDesk.token)
url = URL('https://api.helpdesk.com/v1/').join(endpoint)
_logger.debug(f'POST: {url.copy_merge_params(params)}')
return httpx.post(url, auth=auth, params=params, json=data, headers=self._headers)

def post(self, endpoint: str, data: dict[str, Any], params: dict[str, str] | None = None) -> JSON:
def post(self, endpoint: str, data: dict[str, Any], params: QueryParams | None = None) -> JSON:
resp = self._post_throttled(endpoint, data, params)
resp.raise_for_status()
return resp.json()
Expand Down
2 changes: 1 addition & 1 deletion src/pytanis/helpdesk/mail.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
_logger = get_logger()


class MetaData(BaseModel, extra='allow'): # type: ignore
class MetaData(BaseModel, extra='allow'):
"""Additional, arbitrary metadata provided by the user like for template filling"""


Expand Down
12 changes: 6 additions & 6 deletions src/pytanis/helpdesk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,15 @@ class Id(BaseModel):
ID: str


class Agent(BaseModel, extra='allow'): # type: ignore
class Agent(BaseModel, extra='allow'):
pass


class Team(BaseModel, extra='allow'): # type: ignore
class Team(BaseModel, extra='allow'):
pass


class Message(BaseModel, extra='allow'): # type: ignore
class Message(BaseModel, extra='allow'):
text: str


Expand All @@ -31,12 +31,12 @@ class Requester(BaseModel):
name: str


class Assignment(BaseModel, extra='allow'): # type: ignore
class Assignment(BaseModel, extra='allow'):
team: Id
agent: Id


class NewTicket(BaseModel, extra='allow'): # type: ignore
class NewTicket(BaseModel, extra='allow'):
"""Object that needs to be sent when creating a NEW ticket"""

message: Message
Expand All @@ -47,5 +47,5 @@ class NewTicket(BaseModel, extra='allow'): # type: ignore
assignment: Assignment | None = None


class Ticket(BaseModel, extra='allow'): # type: ignore
class Ticket(BaseModel, extra='allow'):
"""Actual ticket as returned by the API"""
55 changes: 29 additions & 26 deletions src/pytanis/pretalx/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,6 @@
"""Type of a JSON list of JSON objects"""
JSON: TypeAlias = JSONObj | JSONLst
"""Type of the JSON response as returned by the Pretalx API"""
QueryParamType: TypeAlias = dict[Any, Any | list[Any]] | QueryParams
"""Type for the optional parameters to the Pretalx API"""


class PretalxClient:
Expand All @@ -50,15 +48,20 @@ def set_throttling(self, calls: int, seconds: int):
_logger.info('throttling', calls=calls, seconds=seconds)
self._get_throttled = throttle(calls, seconds)(self._get)

def _get(self, endpoint: str, params: QueryParamType | None = None) -> Response:
def _get(self, endpoint: str, params: QueryParams | None = None) -> Response:
"""Retrieve data via GET request"""
auth = HeaderApiKey(self._config.Pretalx.api_token, header_name='Authorization')
if params is None:
params = cast(QueryParams, {})
if (api_token := self._config.Pretalx.api_token) is None:
msg = 'API token for Pretalx is empty'
raise RuntimeError(msg)
auth = HeaderApiKey(api_token, header_name='Authorization')
url = URL('https://pretalx.com/').join(endpoint).copy_merge_params(params)
_logger.info(f'GET: {url}')
# we set the timeout to 60 seconds as the Pretalx API is quite slow
return httpx.get(url, auth=auth, timeout=60.0)

def _get_one(self, endpoint: str, params: QueryParamType | None = None) -> JSON:
def _get_one(self, endpoint: str, params: QueryParams | None = None) -> JSON:
"""Retrieve a single resource result"""
resp = self._get_throttled(endpoint, params)
resp.raise_for_status()
Expand All @@ -73,7 +76,7 @@ def _resolve_pagination(self, resp: JSONObj) -> Iterator[JSONObj]:
_log_resp(resp)
yield from resp['results']

def _get_many(self, endpoint: str, params: QueryParamType | None = None) -> tuple[int, Iterator[JSONObj]]:
def _get_many(self, endpoint: str, params: QueryParams | None = None) -> tuple[int, Iterator[JSONObj]]:
"""Retrieves the result count as well as the results as iterator"""
resp = self._get_one(endpoint, params)
_log_resp(resp)
Expand All @@ -92,7 +95,7 @@ def _endpoint_lst(
event_slug: str,
resource: str,
*,
params: QueryParamType | None = None,
params: QueryParams | None = None,
) -> tuple[int, Iterator[T]]:
"""Queries an endpoint returning a list of resources"""
endpoint = f'/api/events/{event_slug}/{resource}/'
Expand All @@ -107,7 +110,7 @@ def _endpoint_id(
resource: str,
id: int | str, # noqa: A002
*,
params: QueryParamType | None = None,
params: QueryParams | None = None,
) -> T:
"""Query an endpoint returning a single resource"""
endpoint = f'/api/events/{event_slug}/{resource}/{id}/'
Expand All @@ -120,80 +123,80 @@ def me(self) -> Me:
result = self._get_one('/api/me')
return Me.model_validate(result)

def event(self, event_slug: str, *, params: QueryParamType | None = None) -> Event:
def event(self, event_slug: str, *, params: QueryParams | None = None) -> Event:
"""Returns detailed information about a specific event"""
endpoint = f'/api/events/{event_slug}/'
result = self._get_one(endpoint, params)
_logger.debug('result', resp=result)
return Event.model_validate(result)

def events(self, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Event]]:
def events(self, *, params: QueryParams | None = None) -> tuple[int, Iterator[Event]]:
"""Lists all events and their details"""
count, results = self._get_many('/api/events/', params)
events = iter(_logger.debug('result', resp=r) or Event.model_validate(r) for r in results)
return count, events

def submission(self, event_slug: str, code: str, *, params: QueryParamType | None = None) -> Submission:
def submission(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Submission:
"""Returns a specific submission"""
return self._endpoint_id(Submission, event_slug, 'submissions', code, params=params)

def submissions(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Submission]]:
def submissions(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Submission]]:
"""Lists all submissions and their details"""
return self._endpoint_lst(Submission, event_slug, 'submissions', params=params)

def talk(self, event_slug: str, code: str, *, params: QueryParamType | None = None) -> Talk:
def talk(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Talk:
"""Returns a specific talk"""
return self._endpoint_id(Talk, event_slug, 'talks', code, params=params)

def talks(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Talk]]:
def talks(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Talk]]:
"""Lists all talks and their details"""
return self._endpoint_lst(Talk, event_slug, 'talks', params=params)

def speaker(self, event_slug: str, code: str, *, params: QueryParamType | None = None) -> Speaker:
def speaker(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Speaker:
"""Returns a specific speaker"""
return self._endpoint_id(Speaker, event_slug, 'speakers', code, params=params)

def speakers(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Speaker]]:
def speakers(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Speaker]]:
"""Lists all speakers and their details"""
return self._endpoint_lst(Speaker, event_slug, 'speakers', params=params)

def review(self, event_slug: str, id: int, *, params: QueryParamType | None = None) -> Review: # noqa: A002
def review(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Review: # noqa: A002
"""Returns a specific review"""
return self._endpoint_id(Review, event_slug, 'reviews', id, params=params)

def reviews(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Review]]:
def reviews(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Review]]:
"""Lists all reviews and their details"""
return self._endpoint_lst(Review, event_slug, 'reviews', params=params)

def room(self, event_slug: str, id: int, *, params: QueryParamType | None = None) -> Room: # noqa: A002
def room(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Room: # noqa: A002
"""Returns a specific room"""
return self._endpoint_id(Room, event_slug, 'rooms', id, params=params)

def rooms(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Room]]:
def rooms(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Room]]:
"""Lists all rooms and their details"""
return self._endpoint_lst(Room, event_slug, 'rooms', params=params)

def question(self, event_slug: str, id: int, *, params: QueryParamType | None = None) -> Question: # noqa: A002
def question(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Question: # noqa: A002
"""Returns a specific question"""
return self._endpoint_id(Question, event_slug, 'questions', id, params=params)

def questions(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Question]]:
def questions(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Question]]:
"""Lists all questions and their details"""
return self._endpoint_lst(Question, event_slug, 'questions', params=params)

def answer(self, event_slug: str, id: int, *, params: QueryParamType | None = None) -> Answer: # noqa: A002
def answer(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Answer: # noqa: A002
"""Returns a specific answer"""
return self._endpoint_id(Answer, event_slug, 'answers', id, params=params)

def answers(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Answer]]:
def answers(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Answer]]:
"""Lists all answers and their details"""
return self._endpoint_lst(Answer, event_slug, 'answers', params=params)

def tag(self, event_slug: str, tag: str, *, params: QueryParamType | None = None) -> Tag:
def tag(self, event_slug: str, tag: str, *, params: QueryParams | None = None) -> Tag:
"""Returns a specific tag"""
return self._endpoint_id(Tag, event_slug, 'tags', tag, params=params)

def tags(self, event_slug: str, *, params: QueryParamType | None = None) -> tuple[int, Iterator[Tag]]:
def tags(self, event_slug: str, *, params: QueryParams | None = None) -> tuple[int, Iterator[Tag]]:
"""Lists all tags and their details"""
return self._endpoint_lst(Tag, event_slug, 'tags', params=params)

Expand Down
2 changes: 1 addition & 1 deletion src/pytanis/pretalx/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ class Me(BaseModel):
timezone: str


class MultiLingualStr(BaseModel, extra='allow'): # type: ignore
class MultiLingualStr(BaseModel, extra='allow'):
# ToDo: Add here more available languages, not mentioned in the API
en: str | None = None # we assume though that english is always given to simplify things
de: str | None = None
Expand Down

0 comments on commit 990e0f0

Please sign in to comment.