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