Change requests to async

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)

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