feat: add PyPI Stats API client with retries

src/repolytics/ingestion/_http.py

@@ -0,0 +1,19 @@
+"""Shared HTTP helpers for the ingestion API clients."""
+
+import random
+
+import httpx
+
+# HTTP statuses worth retrying.
+RETRYABLE_STATUSES = frozenset({429, 500, 502, 503})
+
+
+def retry_delay(response: httpx.Response, attempt: int, backoff_base: float) -> float:
+    """Seconds to wait before a retry: `Retry-After` if present, else backoff."""
+    retry_after = response.headers.get("Retry-After")
+    if retry_after is not None:
+        try:
+            return float(retry_after)
+        except ValueError:
+            pass
+    return backoff_base * 2**attempt + random.uniform(0, backoff_base)

src/repolytics/ingestion/github_client.py

@@ -1,20 +1,17 @@
 """GitHub REST API client.
 
-Methods return raw JSON exactly as received.
-Handles authentication, pagination, rate limiting, and retries with exponential backoff.
+Methods return raw JSON exactly as received. Handles authentication,
+pagination, rate limiting, and retries with exponential backoff.
 """
 
-import random
 import time
 from collections.abc import Callable
 from datetime import UTC, datetime
 
 import httpx
 
 from repolytics.config import get_settings
-
-# HTTP statuses worth retrying.
-RETRYABLE_STATUSES = frozenset({429, 500, 502, 503})
+from repolytics.ingestion._http import RETRYABLE_STATUSES, retry_delay
 
 
 def _to_iso(dt: datetime) -> str:
@@ -167,22 +164,12 @@ def _request(
         for attempt in range(self._max_retries):
             if response.status_code not in RETRYABLE_STATUSES:
                 break
-            self._sleep(self._retry_delay(response, attempt))
+            self._sleep(retry_delay(response, attempt, self._backoff_base))
             response = self._client.request(method, url, params=params)
         response.raise_for_status()
         self._guard_rate_limit(response)
         return response
 
-    def _retry_delay(self, response: httpx.Response, attempt: int) -> float:
-        """Seconds to wait before a retry: `Retry-After` if present, else backoff."""
-        retry_after = response.headers.get("Retry-After")
-        if retry_after is not None:
-            try:
-                return float(retry_after)
-            except ValueError:
-                pass
-        return self._backoff_base * 2**attempt + random.uniform(0, self._backoff_base)
-
     def _guard_rate_limit(self, response: httpx.Response) -> None:
         """Sleep until the rate-limit window resets when remaining calls run low."""
         remaining = response.headers.get("X-RateLimit-Remaining")

src/repolytics/ingestion/pypi_client.py

@@ -0,0 +1,85 @@
+"""PyPI Stats API client.
+
+Methods return raw JSON exactly as received. Handles retries with exponential
+backoff and keeps a courtesy delay between requests to stay within PyPI's
+informal rate limits.
+"""
+
+import time
+from collections.abc import Callable
+
+import httpx
+
+from repolytics.ingestion._http import RETRYABLE_STATUSES, retry_delay
+
+
+class PyPIClient:
+    """Synchronous client for the PyPI Stats API.
+
+    Wraps `httpx.Client`. Usable as a context manager so the underlying
+    connection pool is closed on exit.
+    """
+
+    def __init__(
+        self,
+        *,
+        base_url: str = "https://pypistats.org/api",
+        min_interval: float = 1.0,
+        max_retries: int = 3,
+        backoff_base: float = 1.0,
+        timeout: httpx.Timeout | None = None,
+        transport: httpx.BaseTransport | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self._min_interval = min_interval
+        self._max_retries = max_retries
+        self._backoff_base = backoff_base
+        self._sleep = sleep
+        self._client = httpx.Client(
+            base_url=base_url,
+            headers={"Accept": "application/json"},
+            timeout=timeout
+            or httpx.Timeout(connect=30.0, read=60.0, write=60.0, pool=60.0),
+            transport=transport,
+        )
+
+    # --- Lifecycle ---
+
+    def close(self) -> None:
+        """Close the underlying HTTP connection pool."""
+        self._client.close()
+
+    def __enter__(self) -> "PyPIClient":
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.close()
+
+    # --- Endpoints ---
+
+    def get_recent_downloads(self, package: str) -> dict:
+        """`GET /packages/{package}/recent` (last day/week/month totals)."""
+        return self._request("GET", f"/packages/{package}/recent").json()
+
+    def get_overall_downloads(self, package: str, *, mirrors: bool = False) -> dict:
+        """`GET /packages/{package}/overall` (daily download time series)."""
+        params = {"mirrors": str(mirrors).lower()}
+        return self._request(
+            "GET", f"/packages/{package}/overall", params=params
+        ).json()
+
+    # --- Internals ---
+
+    def _request(
+        self, method: str, url: str, params: dict[str, object] | None = None
+    ) -> httpx.Response:
+        """Issue one request, retrying transient failures, then pause politely."""
+        response = self._client.request(method, url, params=params)
+        for attempt in range(self._max_retries):
+            if response.status_code not in RETRYABLE_STATUSES:
+                break
+            self._sleep(retry_delay(response, attempt, self._backoff_base))
+            response = self._client.request(method, url, params=params)
+        response.raise_for_status()
+        self._sleep(self._min_interval)
+        return response

src/repolytics/ingestion/watermarks.py

@@ -0,0 +1,30 @@
+"""Watermark store for incremental extraction.
+
+Tracks the last-seen timestamp per `{source}/{endpoint}/{repo}` as a flat JSON
+map on disk, so each ingestion run only fetches records newer than last time.
+"""
+
+import json
+from pathlib import Path
+
+
+def watermark_key(source: str, endpoint: str, repo: str) -> str:
+    """Build the watermark map key for a source/endpoint/repo target."""
+    return f"{source}/{endpoint}/{repo}"
+
+
+def load_watermarks(path: str | Path) -> dict[str, str]:
+    """Load the watermark map, returning an empty dict when the file is absent."""
+    file = Path(path)
+    if not file.exists():
+        return {}
+    return json.loads(file.read_text(encoding="utf-8"))
+
+
+def save_watermarks(watermarks: dict[str, str], path: str | Path) -> None:
+    """Atomically write the watermark map to `path`, creating parents as needed."""
+    file = Path(path)
+    file.parent.mkdir(parents=True, exist_ok=True)
+    tmp = file.with_name(f"{file.name}.tmp")
+    tmp.write_text(json.dumps(watermarks, indent=2), encoding="utf-8")
+    tmp.replace(file)

src/repolytics/ingestion/writer.py

@@ -0,0 +1,45 @@
+"""Parquet writer for raw API responses.
+
+Lands each record as a single JSON `data` column plus a `_loaded_at` timestamp.
+"""
+
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+
+import polars as pl
+
+# Raw-landing schema.
+_SCHEMA = {"data": pl.Utf8, "_loaded_at": pl.Datetime(time_unit="us", time_zone="UTC")}
+
+
+def write_parquet(records: list[dict], path: str | Path) -> Path:
+    """Write `records` to `path` as Parquet (JSON `data` column + `_loaded_at`).
+
+    Each record is serialized to a JSON string and stamped with a single batch
+    timestamp. An empty `records` list still writes a zero-row file with the
+    correct schema, so the partition stays present and schema-stable.
+    """
+    out = Path(path)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    loaded_at = datetime.now(UTC)
+    frame = pl.DataFrame(
+        {
+            "data": [json.dumps(record, ensure_ascii=False) for record in records],
+            "_loaded_at": [loaded_at] * len(records),
+        },
+        schema=_SCHEMA,
+    )
+    frame.write_parquet(out)
+    return out
+
+
+def partition_path(
+    root: str | Path, source: str, table: str, date: datetime | str
+) -> Path:
+    """Build the date-partitioned landing path for a source/table on a date.
+
+    Returns `{root}/{source}/{table}/{YYYY-MM-DD}.parquet`.
+    """
+    day = date.strftime("%Y-%m-%d") if isinstance(date, datetime) else date
+    return Path(root) / source / table / f"{day}.parquet"

tests/unit/_clients.py

@@ -0,0 +1,27 @@
+"""Builders for mock-transport-backed API clients."""
+
+from collections.abc import Callable
+
+import httpx
+
+
+def mock_client(
+    client_cls: type,
+    handler: Callable[[httpx.Request], httpx.Response],
+    *args: object,
+    **kwargs: object,
+) -> tuple[object, list[float]]:
+    """Build an ingestion client over a `MockTransport` with a spy `sleep`.
+
+    Returns the client and the list recording every sleep duration, so tests can
+    assert on retry/courtesy waits without blocking.
+    """
+    sleeps: list[float] = []
+    client = client_cls(
+        *args,
+        transport=httpx.MockTransport(handler),
+        sleep=sleeps.append,
+        backoff_base=0.0,
+        **kwargs,
+    )
+    return client, sleeps

tests/unit/test_github_client.py

@@ -7,7 +7,9 @@
 import httpx
 import pytest
 
+from repolytics.ingestion._http import RETRYABLE_STATUSES
 from repolytics.ingestion.github_client import GitHubClient, _to_iso
+from tests.unit._clients import mock_client
 
 # Absolute next-page URL used to drive pagination in tests.
 NEXT_URL = "https://api.github.com/x?page=2"
@@ -17,25 +19,13 @@
 def make_client(
     handler: Callable[[httpx.Request], httpx.Response], **kwargs: object
 ) -> tuple[GitHubClient, list[float]]:
-    """Build a client backed by a `MockTransport` and a spy `sleep`.
-
-    Returns the client and the list that records every sleep duration, so tests
-    can assert on retry/rate-limit waits without ever blocking.
-    """
-    sleeps: list[float] = []
-    client = GitHubClient(
-        "test-token",
-        transport=httpx.MockTransport(handler),
-        sleep=sleeps.append,
-        backoff_base=0.0,
-        **kwargs,
-    )
-    return client, sleeps
+    """Build a `GitHubClient` over a `MockTransport` with a spy `sleep`."""
+    return mock_client(GitHubClient, handler, "test-token", **kwargs)
 
 
-def test_to_iso_appends_z_for_naive_and_aware() -> None:
-    assert _to_iso(datetime(2024, 1, 1)) == "2024-01-01T00:00:00Z"
-    assert _to_iso(datetime(2024, 1, 1, tzinfo=UTC)) == "2024-01-01T00:00:00Z"
+@pytest.mark.parametrize("dt", [datetime(2024, 1, 1), datetime(2024, 1, 1, tzinfo=UTC)])
+def test_to_iso_appends_z(dt: datetime) -> None:
+    assert _to_iso(dt) == "2024-01-01T00:00:00Z"
 
 
 def test_paginates_following_next_link() -> None:
@@ -77,23 +67,24 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert sleeps[0] > 0
 
 
-def test_retries_then_succeeds() -> None:
-    statuses = [502, 502, 200]
+@pytest.mark.parametrize("status", sorted(RETRYABLE_STATUSES))
+def test_retryable_status_retries_then_succeeds(status: int) -> None:
+    statuses = [status, 200]
     calls: list[httpx.Request] = []
 
     def handler(request: httpx.Request) -> httpx.Response:
-        status = statuses[len(calls)]
+        current = statuses[len(calls)]
         calls.append(request)
-        if status == 200:
+        if current == 200:
             return httpx.Response(200, json={"id": 1})
-        return httpx.Response(status)
+        return httpx.Response(current)
 
     client, sleeps = make_client(handler)
     result = client.get_repository("o", "r")
 
     assert result == {"id": 1}
-    assert len(calls) == 3
-    assert len(sleeps) == 2
+    assert len(calls) == 2  # one failure, one retry
+    assert len(sleeps) == 1
 
 
 def test_raises_after_max_retries() -> None:
@@ -107,12 +98,13 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert len(sleeps) == 3
 
 
-def test_404_raises_without_retry() -> None:
+@pytest.mark.parametrize("status", [400, 401, 403, 404])
+def test_non_retryable_status_raises_without_retry(status: int) -> None:
     calls: list[httpx.Request] = []
 
     def handler(request: httpx.Request) -> httpx.Response:
         calls.append(request)
-        return httpx.Response(404)
+        return httpx.Response(status)
 
     client, sleeps = make_client(handler)
     with pytest.raises(httpx.HTTPStatusError):