cryptocurrencydataapiservice-zprlcd/app/services/coincap_api.py

import httpx
from typing import Dict, List, Optional, Any, Union
from datetime import datetime
from fastapi import HTTPException

from app.core.config import settings


class CoinCapAPI:
    """
    Service class for interacting with the CoinCap API
    """
    
    def __init__(self):
        self.base_url = settings.COINCAP_API_URL
        self.api_key = settings.COINCAP_API_KEY
        self.headers = {"Authorization": f"Bearer {self.api_key}"}
    
    async def _make_request(
        self, 
        endpoint: str, 
        params: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Make an HTTP request to the CoinCap API
        """
        url = f"{self.base_url}{endpoint}"
        
        # Add API key to params
        if params is None:
            params = {}
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    url, 
                    params=params,
                    headers=self.headers,
                    timeout=10.0
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 404:
                    raise HTTPException(status_code=404, detail="Resource not found")
                else:
                    raise HTTPException(
                        status_code=response.status_code, 
                        detail=f"CoinCap API error: {response.text}"
                    )
        except httpx.RequestError as e:
            raise HTTPException(status_code=503, detail=f"Could not connect to CoinCap API: {str(e)}")
    
    # Assets endpoints
    async def get_assets(
        self, 
        search: Optional[str] = None, 
        ids: Optional[str] = None, 
        limit: Optional[int] = 100, 
        offset: Optional[int] = 0
    ) -> Dict[str, Any]:
        """
        Get a list of assets with optional filters
        """
        params = {
            "limit": limit,
            "offset": offset
        }
        
        if search:
            params["search"] = search
        if ids:
            params["ids"] = ids
            
        return await self._make_request("/assets", params)
    
    async def get_asset(self, asset_id: str) -> Dict[str, Any]:
        """
        Get details for a specific asset by ID (slug)
        """
        return await self._make_request(f"/assets/{asset_id}")
    
    async def get_asset_markets(
        self, 
        asset_id: str, 
        limit: Optional[int] = 100, 
        offset: Optional[int] = 0
    ) -> Dict[str, Any]:
        """
        Get markets for a specific asset
        """
        params = {
            "limit": limit,
            "offset": offset
        }
        
        return await self._make_request(f"/assets/{asset_id}/markets", params)
    
    async def get_asset_history(
        self,
        asset_id: str,
        interval: str,
        start: Optional[int] = None,
        end: Optional[int] = None
    ) -> Dict[str, Any]:
        """
        Get historical data for a specific asset
        Interval choices: m1, m5, m15, m30, h1, h2, h6, h12, d1
        """
        params = {
            "interval": interval
        }
        
        if start is not None and end is not None:
            params["start"] = start
            params["end"] = end
            
        return await self._make_request(f"/assets/{asset_id}/history", params)
    
    # Exchanges endpoints
    async def get_exchanges(
        self, 
        limit: Optional[int] = 10, 
        offset: Optional[int] = 0
    ) -> Dict[str, Any]:
        """
        Get a list of exchanges
        """
        params = {
            "limit": limit,
            "offset": offset
        }
        
        return await self._make_request("/exchanges", params)
    
    async def get_exchange(self, exchange_id: str) -> Dict[str, Any]:
        """
        Get details for a specific exchange
        """
        return await self._make_request(f"/exchanges/{exchange_id}")
    
    # Markets endpoints
    async def get_markets(
        self,
        exchange_id: Optional[str] = None,
        base_symbol: Optional[str] = None,
        base_id: Optional[str] = None,
        quote_symbol: Optional[str] = None,
        quote_id: Optional[str] = None,
        asset_symbol: Optional[str] = None,
        asset_id: Optional[str] = None,
        limit: Optional[int] = 10,
        offset: Optional[int] = 0
    ) -> Dict[str, Any]:
        """
        Get a list of markets with optional filters
        """
        params = {
            "limit": limit,
            "offset": offset
        }
        
        if exchange_id:
            params["exchangeId"] = exchange_id
        if base_symbol:
            params["baseSymbol"] = base_symbol
        if base_id:
            params["baseId"] = base_id
        if quote_symbol:
            params["quoteSymbol"] = quote_symbol
        if quote_id:
            params["quoteId"] = quote_id
        if asset_symbol:
            params["assetSymbol"] = asset_symbol
        if asset_id:
            params["assetId"] = asset_id
            
        return await self._make_request("/markets", params)
    
    # Rates endpoints
    async def get_rates(self, ids: Optional[str] = None) -> Dict[str, Any]:
        """
        Get a list of all conversion rates or filter by IDs
        """
        params = {}
        if ids:
            params["ids"] = ids
            
        return await self._make_request("/rates", params)
    
    async def get_rate(self, rate_id: str) -> Dict[str, Any]:
        """
        Get a specific conversion rate by ID
        """
        return await self._make_request(f"/rates/{rate_id}")


coincap_api = CoinCapAPI()