perf(http): Shift from requests to asyncio/aiohttp

my_chess_style/settings/base.py

@@ -141,10 +141,9 @@
 celery_pass = os.getenv("RABBITMQ_DEFAULT_PASS")
 net_host = os.getenv("DB_HOST")
 
+REDIS_URI = f"redis://:{os.getenv('CACHE_PASSWORD')}@{os.getenv('DB_HOST')}:6379/0"
 CELERY_BROKER_URL = f"amqp://{celery_user}:{celery_pass}@{net_host}:5672/"
-CELERY_RESULT_BACKEND = (
-    f"redis://:{os.getenv('CACHE_PASSWORD')}@{os.getenv('DB_HOST')}:6379/0"
-)
+CELERY_RESULT_BACKEND = REDIS_URI
 CELERY_RESULT_SERIALIZER = "json"
 CELERY_ACCEPT_CONTENT = ["json", "pickle"]
 CELERY_RESULT_ACCEPT_CONTENT = ["json", "pickle"]
@@ -154,6 +153,6 @@
 CACHES = {
     "default": {
         "BACKEND": "django.core.cache.backends.redis.RedisCache",
-        "LOCATION": f"redis://:{os.getenv('CACHE_PASSWORD')}@{os.getenv('DB_HOST')}:6379",
+        "LOCATION": f"redis://default:{os.getenv('CACHE_PASSWORD')}@{os.getenv('DB_HOST')}:6379/1",
     }
 }

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
   "langchain-ollama>=0.3.6",
   "matplotlib>=3.10.1",
   "numpy>=2.2.4",
+  "palitra>=0.0.1.post1",
   "pandas>=2.2.3",
   "psycopg2>=2.9.10",
   "pyarrow>=20.0.0",

requirements.txt

@@ -542,6 +542,10 @@ packaging==24.2 \
     #   langsmith
     #   matplotlib
     #   pytest
+palitra==0.0.1.post1 \
+    --hash=sha256:9faac26fc232706b99a87823b82cf5f63c533576799b17d6c835a49d0e2539de \
+    --hash=sha256:a26a67e0ffe16022d4108490a674634ae5c42c555c40edcf2c863d9795dcad7b
+    # via my-chess-style
 pandas==2.2.3 \
     --hash=sha256:15c0e1e02e93116177d29ff83e8b1619c93ddc9c49083f237d4312337a61165d \
     --hash=sha256:1db71525a1538b30142094edb9adc10be3f3e176748cd7acc2240c2f2e5aa3a4 \

style_predictor/apis/pgn/utils.py

@@ -1,10 +1,11 @@
+import asyncio
 import logging
 import os
-import time
 
+import aiohttp
 import berserk
-import requests
 from chessdotcom import ChessDotComClient
+from django.core.cache import cache
 from dotenv import load_dotenv
 
 LOG = logging.getLogger(__name__)
@@ -38,13 +39,21 @@ def does_lichess_player_exists(username: str):
     ResponseError: if user is not Found
 
     """
-    LOG.info(f"Checking if {username} exists on Lichess")
-    try:
-        _ = lichessClient.users.get_public_data(username)
-    except Exception as e:
-        LOG.error(e)
-        return False
-    return True
+    cache_key: str = f"lichess-{username}"
+    is_present: bool | None = cache.get(cache_key)
+    if is_present is None:
+        LOG.info(f"Checking if {username} exists on Lichess")
+        try:
+            _ = lichessClient.users.get_public_data(username)
+            cache.set(
+                cache_key,
+                True,
+            )
+        except Exception as e:
+            LOG.error(e)
+            return False
+        return True
+    return is_present
 
 
 def get_lichess_games(username: str) -> str:
@@ -76,16 +85,44 @@ def does_chess_dot_com_player_exists(username: str):
     Raises:
     ChessDotComClientError: if user is not Found
     """
+    cache_key: str = f"chess_dot_com-{username}"
     LOG.info(f"Checking if {username} exists on Chess.com")
-    try:
-        _ = chessdotcomClient.get_player_profile(username)
-    except Exception as e:
-        LOG.error(e)
-        return False
-    return True
-
-
-def get_chess_dot_com_games(username: str) -> str:
+    is_present: bool | None = cache.get(cache_key)
+    if is_present is None:
+        try:
+            _ = chessdotcomClient.get_player_profile(username)
+            cache.set(
+                cache_key,
+                True,
+            )
+        except Exception as e:
+            LOG.error(e)
+            return False
+        return True
+    return is_present
+
+
+async def fetch_archive(
+    archive_url: str, session: aiohttp.ClientSession, semaphore: asyncio.Semaphore
+):
+    async with semaphore:
+        while True:
+            async with session.get(f"{archive_url}/pgn") as resp:
+                if resp.status == 429:
+                    if retry_after := resp.headers.get("Retry-After"):
+                        LOG.info(f"Rate limited. Retrying after {retry_after} seconds.")
+                        await asyncio.sleep(int(retry_after))
+                    else:
+                        # Implement exponential backoff here if no Retry-After
+                        await asyncio.sleep(45)  # Example fixed delay
+                elif resp.status == 200:
+                    return await resp.text()
+                else:
+                    LOG.error(f"Failed to fetch {archive_url}: {resp.status}")
+                    return ""
+
+
+async def get_chess_dot_com_games(username: str) -> str:
     """Get all games of `username` from chess.com platform.
 
     Args:
@@ -94,19 +131,16 @@ def get_chess_dot_com_games(username: str) -> str:
     Returns:
         string of all games.
     """
-    all_pgns: str = ""
+    conn_limit = 15
+    semaphore = asyncio.Semaphore(conn_limit)
+    all_pgns: list[str] = []
     response = chessdotcomClient.get_player_game_archives(username)
     archives: dict[str, list[str]] = response.json
-    for archive in archives.get("archives", []):
-        resp = requests.get(f"{archive}/pgn", headers=headers, timeout=60)
-        if resp.status_code == 200:
-            all_pgns += f"{resp.text}\n\n"
-        elif resp.status_code == 429:
-            LOG.warning(f"Rate limiting encountered for {archive}")
-            time.sleep(45)
-            resp = requests.get(f"{archive}/pgn", headers=headers, timeout=60)
-            if resp.status_code == 200:
-                all_pgns += f"{resp.text}\n\n"
-            else:
-                LOG.error(f"Failed to fetch url {archive}/pgn")
-    return all_pgns
+    connection = aiohttp.TCPConnector(limit=conn_limit)
+    async with aiohttp.ClientSession(connector=connection) as session:
+        tasks = [
+            fetch_archive(archive, session, semaphore)
+            for archive in archives.get("archives", [])
+        ]
+        all_pgns = await asyncio.gather(*tasks)
+    return "\n\n".join([pgn for pgn in all_pgns if pgn])

style_predictor/tasks.py

@@ -4,6 +4,7 @@
 from typing import Any, Callable, NamedTuple
 from uuid import UUID
 
+import palitra
 from celery import chord, current_app, shared_task
 from celery.signals import task_postrun
 from django.core.cache import cache
@@ -311,7 +312,7 @@ def pgn_get_games_from_file(session_id: UUID, usernames: str, pgn_data: str):
 @shared_task(name=constants.GET_CHESS_COM_TASK)
 def pgn_get_chess_com_games_by_user(session_id: UUID, username: str):
     """Celery task to get chess games for user from chess.com."""
-    pgn_data: str = get_chess_dot_com_games(username)
+    pgn_data: str = palitra.run(get_chess_dot_com_games(username))
     return save_file_and_queue_task(
         session_id, username, pgn_data, FileSource.CHESSDOTCOM
     )