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

We have better performance switching to aiohttp.

Author: CalvoM

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"]

style_predictor/apis/pgn/utils.py

@@ -1,9 +1,9 @@
+import asyncio
 import logging
 import os
-import time
 
+import aiohttp
 import berserk
-import requests
 from chessdotcom import ChessDotComClient
 from dotenv import load_dotenv
 
@@ -85,7 +85,28 @@ def does_chess_dot_com_player_exists(username: str):
     return True
 
 
-def get_chess_dot_com_games(username: str) -> str:
+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:
+                    retry_after = resp.headers.get("Retry-After")
+                    if 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 +115,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(all_pgns)

style_predictor/tasks.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from collections import Counter, defaultdict
 from dataclasses import dataclass
@@ -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 = asyncio.run(get_chess_dot_com_games(username))
     return save_file_and_queue_task(
         session_id, username, pgn_data, FileSource.CHESSDOTCOM
     )