diff --git a/src/cloud_util/homeassistant.py b/src/cloud_util/homeassistant.py index 5b98bee..c2643aa 100644 --- a/src/cloud_util/homeassistant.py +++ b/src/cloud_util/homeassistant.py @@ -19,9 +19,9 @@ class HomeAssistant: def __init__(self, ticktick: TickTick) -> None: self._ticktick = ticktick - def process_message(self, message: Message) -> dict[str, str]: + async def process_message(self, message: Message) -> dict[str, str]: if message.target == "ticktick": - return self._process_ticktick_message(message=message) + return await self._process_ticktick_message(message=message) return {"Status": "Unknown target"} @@ -31,21 +31,21 @@ class HomeAssistant: headers: dict[str, str] = {"Authorization": f"Bearer {token}"} await httpx.AsyncClient().post(webhook_url, json=payload, headers=headers) - def _process_ticktick_message(self, message: Message) -> dict[str, str]: + async def _process_ticktick_message(self, message: Message) -> dict[str, str]: if message.action == "create_shopping_list": - return self._create_shopping_list(content=message.content) + return await self._create_shopping_list(content=message.content) if message.action == "create_action_task": - return self._create_action_task(content=message.content) + return await self._create_action_task(content=message.content) return {"Status": "Unknown action"} - def _create_shopping_list(self, content: str) -> dict[str, str]: + async def _create_shopping_list(self, content: str) -> dict[str, str]: project_id = Config.get_env("TICKTICK_SHOPPING_LIST") item: dict[str, str] = ast.literal_eval(content) task = TickTick.Task(projectId=project_id, title=item["item"]) - return self._ticktick.create_task(task=task) + return await self._ticktick.create_task(task=task) - def _create_action_task(self, content: str) -> dict[str, str]: + async def _create_action_task(self, content: str) -> dict[str, str]: detail: dict[str, str] = ast.literal_eval(content) project_id = Config.get_env("TICKTICK_HOME_TASK_LIST") due_hour = detail["due_hour"] @@ -53,4 +53,4 @@ class HomeAssistant: due = (due + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) due = due.astimezone(timezone.utc) task = TickTick.Task(projectId=project_id, title=detail["action"], dueDate=TickTick.datetime_to_ticktick_format(due)) - return self._ticktick.create_task(task=task) + return await self._ticktick.create_task(task=task) diff --git a/src/cloud_util/ticktick.py b/src/cloud_util/ticktick.py index f471b83..7ef6245 100644 --- a/src/cloud_util/ticktick.py +++ b/src/cloud_util/ticktick.py @@ -7,7 +7,7 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from datetime import datetime -import requests +import httpx from src.config import Config @@ -40,12 +40,12 @@ class TickTick: ticktick_auth_url_encoded = urllib.parse.urlencode(ticktick_code_auth_params) print("Visit: ", ticktick_code_auth_url + ticktick_auth_url_encoded, " to authenticate.") - def retrieve_access_token(self, code: str, state: str) -> bool: + async def retrieve_access_token(self, code: str, state: str) -> bool: if state != "begin_auth": print("Invalid state.") return False ticktick_token_url = "https://ticktick.com/oauth/token" # noqa: S105 - ticktick_token_auth_params = { + ticktick_token_auth_params: dict[str, str] = { "code": code, "grant_type": "authorization_code", "scope": "tasks:write tasks:read", @@ -53,26 +53,30 @@ class TickTick: } client_id = Config.get_env("TICKTICK_CLIENT_ID") client_secret = Config.get_env("TICKTICK_CLIENT_SECRET") - response = requests.post(ticktick_token_url, data=ticktick_token_auth_params, auth=(client_id, client_secret), timeout=10) - print(response) - Config.update_env("TICKTICK_ACCESS_TOKEN", response.json().get("access_token")) + response = await httpx.AsyncClient().post( + ticktick_token_url, + data=ticktick_token_auth_params, + auth=httpx.BasicAuth(username=client_id, password=client_secret), + timeout=10, + ) + Config.update_env("TICKTICK_ACCESS_TOKEN", response.json()["access_token"]) return True - def get_tasks(self, project_id: str) -> list[dict]: + async def get_tasks(self, project_id: str) -> list[dict]: ticktick_get_tasks_url = "https://api.ticktick.com/open/v1/project/" + project_id + "/data" header: dict[str, str] = {"Authorization": f"Bearer {self._access_token}"} - response = requests.get(ticktick_get_tasks_url, headers=header, timeout=10) + response = await httpx.AsyncClient().get(ticktick_get_tasks_url, headers=header, timeout=10) return response.json()["tasks"] - def has_duplicate_task(self, project_id: str, task_title: str) -> bool: - tasks = self.get_tasks(project_id=project_id) + async def has_duplicate_task(self, project_id: str, task_title: str) -> bool: + tasks = await self.get_tasks(project_id=project_id) return any(task["title"] == task_title for task in tasks) - def create_task(self, task: TickTick.Task) -> dict[str, str]: - if not self.has_duplicate_task(project_id=task.projectId, task_title=task.title): + async def create_task(self, task: TickTick.Task) -> dict[str, str]: + if not await self.has_duplicate_task(project_id=task.projectId, task_title=task.title): ticktick_task_creation_url = "https://api.ticktick.com/open/v1/task" header: dict[str, str] = {"Authorization": f"Bearer {self._access_token}"} - requests.post(ticktick_task_creation_url, headers=header, json=asdict(task), timeout=10) + await httpx.AsyncClient().post(ticktick_task_creation_url, headers=header, json=asdict(task), timeout=10) return {"title": task.title} @staticmethod diff --git a/src/main.py b/src/main.py index 64ef304..7df80da 100644 --- a/src/main.py +++ b/src/main.py @@ -40,7 +40,7 @@ async def get_status() -> dict: @app.post("/homeassistant/publish") async def homeassistant_publish(payload: HomeAssistant.Message) -> dict: - return homeassistant.process_message(message=payload) + return await homeassistant.process_message(message=payload) # Poo recorder @@ -53,6 +53,6 @@ async def record(record_detail: PooRecordField) -> PooRecordField: # ticktick @app.get("/ticktick/auth/code") async def ticktick_auth(code: str, state: str) -> dict: - if ticktick.retrieve_access_token(code, state): + if await ticktick.retrieve_access_token(code, state): return {"State": "Token Retrieved"} return {"State": "Token Retrieval Failed"}