Zeta Terminal — New Data Source Integration

Existing Sources (reference patterns)

Service Template

backend/src/services/{source}_service.py:

python
1"""
2{Source Name} data service.
3
4API docs: {url}
5Rate limits: {X req/s}
6Auth: {method}
7"""
8import logging
9import aiohttp
10from typing import Any
11
12logger = logging.getLogger(__name__)
13
14# Constants
15BASE_URL = "https://api.example.com"
16TIMEOUT = aiohttp.ClientTimeout(total=30)
17MAX_RETRIES = 3
18
19class {Source}Service:
20    """Async client for {Source} API."""
21
22    def __init__(self, api_key: str | None = None):
23        self.api_key = api_key
24        self._session: aiohttp.ClientSession | None = None
25
26    async def _get_session(self) -> aiohttp.ClientSession:
27        if self._session is None or self._session.closed:
28            headers = {"Content-Type": "application/json"}
29            if self.api_key:
30                headers["Authorization"] = f"Bearer {self.api_key}"
31            self._session = aiohttp.ClientSession(
32                base_url=BASE_URL,
33                headers=headers,
34                timeout=TIMEOUT,
35            )
36        return self._session
37
38    async def fetch_data(self, params: dict[str, Any]) -> dict[str, Any]:
39        """
40        Fetch data from {Source}.
41
42        Parameters
43        ----------
44        params : dict
45            Query parameters.
46
47        Returns
48        -------
49        dict with keys: data, metadata
50
51        Raises
52        ------
53        ValueError
54            If API returns an error or invalid data.
55        RuntimeError
56            If connection fails after retries.
57        """
58        session = await self._get_session()
59
60        for attempt in range(MAX_RETRIES):
61            try:
62                async with session.get("/endpoint", params=params) as resp:
63                    if resp.status != 200:
64                        text = await resp.text()
65                        logger.error("{Source} API error %d: %s", resp.status, text)
66                        if resp.status == 429:  # Rate limited
67                            await asyncio.sleep(2 ** attempt)
68                            continue
69                        raise ValueError(f"{Source} API returned {resp.status}")
70
71                    data = await resp.json()
72                    return self._parse_response(data)
73            except aiohttp.ClientError as e:
74                logger.error("{Source} connection error (attempt %d): %s", attempt + 1, e)
75                if attempt == MAX_RETRIES - 1:
76                    raise RuntimeError(f"{Source} unavailable after {MAX_RETRIES} retries") from e
77
78    def _parse_response(self, raw: dict) -> dict[str, Any]:
79        """Parse and validate API response."""
80        # Transform to standard format
81        return {
82            "data": raw,
83            "metadata": {
84                "source": "{source}",
85                "timestamp": datetime.now(datetime.UTC).isoformat(),
86            }
87        }
88
89    async def close(self):
90        if self._session and not self._session.closed:
91            await self._session.close()

Router Pattern

python
1@router.get("/api/{source}/fetch")
2async def fetch_data(
3    param: str = Query(..., description="Query parameter"),
4):
5    service = {Source}Service(api_key=os.environ.get("{SOURCE}_API_KEY"))
6    try:
7        result = await service.fetch_data({"param": param})
8        return result
9    except ValueError as e:
10        raise HTTPException(status_code=400, detail=str(e)) from e
11    except RuntimeError as e:
12        raise HTTPException(status_code=503, detail="Data source unavailable") from e
13    finally:
14        await service.close()

Data Source Priorities (Aladdin Data Platform roadmap)

Checklist Before Done

• Async client with aiohttp (not requests)
• Retry logic with exponential backoff
• Rate limiting respected (429 handling)
• API key from os.environ, never hardcoded
• Timeout configured (aiohttp.ClientTimeout)
• Session cleanup (close() method)
• Response parsed into standard format (data + metadata)
• Router with proper error handling (400/503)
• API docs URL in docstring
• ruff check passes