Update app/price_fetcher.py

app/price_fetcher.py

@@ -1,70 +1,175 @@
 """
-Background price cache with multi-API fallback and rate-limit handling
+Provides a robust, asynchronous PriceFetcher class for caching cryptocurrency prices.
+
+Features:
+- Asynchronous fetching using httpx.AsyncClient.
+- Multi-API fallback for high availability.
+- Rate-limit (429) and error handling.
+- Concurrency-safe in-memory cache.
+- Decoupled design for easy extension with new data sources.
 """
-import httpx
+import asyncio
 import logging
+from typing import Callable, TypedDict, Awaitable
 
-# Primary and secondary APIs for crypto prices
-COINGECKO_URL = (
-    "https://api.coingecko.com/api/v3/simple/price"
-    "?ids=bitcoin,ethereum,dogecoin&vs_currencies=usd"
-)
-COINCAP_URL = (
-    "https://api.coincap.io/v2/assets?ids=bitcoin,ethereum,dogecoin"
+import httpx
+
+# --- Configuration ---
+# Set up a structured logger
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
 )
 
-# Shared price cache
-CURRENT_PRICES = {"bitcoin": "--", "ethereum": "--", "dogecoin": "--"}
+# Define the structure for a price parsing function
+PriceParser = Callable[[dict], Awaitable[dict[str, float]]]
+
+# Define the structure for a data source, linking a URL to its parser
+class PriceSource(TypedDict):
+    name: str
+    url: str
+    params: dict
+    parser: PriceParser
+
 
-# How often to retry each API before falling back (seconds)
-RETRY_DELAY = 5
+# --- Main Class: PriceFetcher ---
+class PriceFetcher:
+    """Manages fetching and caching crypto prices from multiple APIs asynchronously."""
 
+    def __init__(self, client: httpx.AsyncClient, coins: list[str]):
+        """
+        Initializes the PriceFetcher.
 
-def fetch_prices() -> None:
-    """
-    Try CoinGecko first; on 429 or error, fall back to CoinCap.
-    Updates CURRENT_PRICES in-place.
-    """
-    global CURRENT_PRICES
-    apis = [
-        ("CoinGecko", COINGECKO_URL),
-        ("CoinCap", COINCAP_URL)
-    ]
+        Args:
+            client: An instance of httpx.AsyncClient for making API calls.
+            coins: A list of coin IDs to fetch (e.g., ['bitcoin', 'ethereum']).
+        """
+        self.client = client
+        self.coins = coins
+        self._prices: dict[str, float | str] = {coin: "--" for coin in coins}
+        self._lock = asyncio.Lock()  # Lock to prevent race conditions on the cache
+        self.sources: list[PriceSource] = self._configure_sources()
 
-    for name, url in apis:
+    def _configure_sources(self) -> list[PriceSource]:
+        """Defines the API sources and their parsers."""
+        return [
+            {
+                "name": "CoinGecko",
+                "url": "https://api.coingecko.com/api/v3/simple/price",
+                "params": {
+                    "ids": ",".join(self.coins),
+                    "vs_currencies": "usd"
+                },
+                "parser": self._parse_coingecko,
+            },
+            {
+                "name": "CoinCap",
+                "url": "https://api.coincap.io/v2/assets",
+                "params": {"ids": ",".join(self.coins)},
+                "parser": self._parse_coincap,
+            },
+        ]
+
+    async def _parse_coingecko(self, data: dict) -> dict[str, float]:
+        """Parses the JSON response from CoinGecko."""
         try:
-            resp = httpx.get(url, timeout=10)
-            # Handle CoinCap JSON structure separately
-            if name == "CoinGecko":
-                resp.raise_for_status()
-                data = resp.json()
-                prices = {
-                    "bitcoin": data["bitcoin"]["usd"],
-                    "ethereum": data["ethereum"]["usd"],
-                    "dogecoin": data["dogecoin"]["usd"]
-                }
-            else:
-                resp.raise_for_status()
-                data = resp.json().get("data", [])
-                prices = {item["id"]: float(item["priceUsd"]) for item in data}
-
-            CURRENT_PRICES.update(prices)
-            logging.info("✅ [%s] prices updated: %s", name, prices)
-            return
-
-        except httpx.HTTPStatusError as e:
-            status = e.response.status_code
-            if status == 429:
-                logging.warning("⚠️ [%s] rate limit (429). Retrying fallback.", name)
-            else:
-                logging.warning("⚠️ [%s] HTTP error %s: %s", name, status, e)
-        except Exception as e:
-            logging.warning("⚠️ [%s] fetch error: %s", name, e)
-
-        # If we reach here, wait a bit before trying next API
+            return {
+                coin: float(data[coin]["usd"])
+                for coin in self.coins if coin in data
+            }
+        except (KeyError, TypeError) as e:
+            logging.error("❌ [CoinGecko] Failed to parse response: %s", e)
+            return {}
+
+    async def _parse_coincap(self, data: dict) -> dict[str, float]:
+        """Parses the JSON response from CoinCap."""
         try:
-            import time; time.sleep(RETRY_DELAY)
-        except Exception:
-            pass
+            # CoinCap returns a list under the 'data' key
+            return {
+                item["id"]: float(item["priceUsd"])
+                for item in data.get("data", []) if item.get("id") in self.coins
+            }
+        except (KeyError, TypeError, ValueError) as e:
+            logging.error("❌ [CoinCap] Failed to parse response: %s", e)
+            return {}
+
+    def get_current_prices(self) -> dict[str, float | str]:
+        """Returns a copy of the current price cache. Thread-safe read."""
+        return self._prices.copy()
+
+    async def update_prices_async(self) -> None:
+        """
+        Asynchronously fetches prices, trying each source until one succeeds.
+        Updates the internal price cache in a concurrency-safe manner.
+        """
+        for source in self.sources:
+            name, url, params, parser = source.values()
+            try:
+                resp = await self.client.get(url, params=params, timeout=10)
+                resp.raise_for_status()
+                
+                new_prices = await parser(resp.json())
+
+                if not new_prices:  # Parser failed to extract data
+                    continue
+
+                async with self._lock:
+                    self._prices.update(new_prices)
+                
+                logging.info("✅ [%s] Prices updated: %s", name, new_prices)
+                return  # Success, so we exit the loop
+
+            except httpx.HTTPStatusError as e:
+                status = e.response.status_code
+                log_msg = f"⚠️ [{name}] HTTP error {status}"
+                if status == 429:
+                    log_msg += " (Rate Limit). Trying next source..."
+                logging.warning(log_msg)
+
+            except (httpx.RequestError, asyncio.TimeoutError) as e:
+                logging.warning("⚠️ [%s] Request failed: %s. Trying next source...", name, e)
+            
+            # Brief pause before trying the next API source
+            await asyncio.sleep(1)
+
+        logging.error("❌ All price APIs failed. Retaining stale prices.")
+
+
+async def run_price_updates_periodically(fetcher: PriceFetcher, interval_seconds: int):
+    """A background task runner to keep prices updated."""
+    logging.info("🚀 Starting periodic price updates...")
+    while True:
+        await fetcher.update_prices_async()
+        await asyncio.sleep(interval_seconds)
+
+
+# --- Example Usage ---
+if __name__ == "__main__":
+    async def main():
+        """Demonstrates how to use the PriceFetcher."""
+        target_coins = ["bitcoin", "ethereum", "dogecoin"]
+        
+        async with httpx.AsyncClient() as client:
+            price_fetcher = PriceFetcher(client, coins=target_coins)
+
+            # Run the price updates in the background
+            update_task = asyncio.create_task(
+                run_price_updates_periodically(price_fetcher, interval_seconds=10)
+            )
+
+            # In a real app, the server would be running. Here, we just print prices.
+            for i in range(5):
+                await asyncio.sleep(11)
+                current_prices = price_fetcher.get_current_prices()
+                print(f"--- Main App Reading Cache ({i+1}/5) ---")
+                for coin, price in current_prices.items():
+                    print(f"  {coin.capitalize()}: ${price}")
+            
+            # Cleanly shut down the background task
+            update_task.cancel()
+            try:
+                await update_task
+            except asyncio.CancelledError:
+                logging.info("Shutdown complete.")
 
-    logging.error("❌ All price APIs failed. Keeping previous prices: %s", CURRENT_PRICES)
+    asyncio.run(main())