Stage 158 (α1): Stripe customer-health connector

infra/ingestion/connectors.py

@@ -177,10 +177,22 @@ def build_connector(connector_type: str, config: dict) -> Connector:
             api_version=config.get("api_version", "v60.0"),
             timeout=float(config.get("timeout", 30.0)),
         )
+    # Stage 158 (α1) — Stripe customer-health connector.
+    from .stripe_connector import StripeConnector
+    if connector_type == StripeConnector.type_name:
+        # api_key OR api_key_env required; the constructor enforces.
+        return StripeConnector(
+            api_key=config.get("api_key"),
+            api_key_env=config.get("api_key_env"),
+            limit=int(config.get("limit", 100)),
+            timeout=float(config.get("timeout", 30.0)),
+            base_url=config.get("base_url", "https://api.stripe.com"),
+        )
     raise ValueError(
         f"unknown / non-schedulable connector type {connector_type!r} "
         f"(schedulable: ['{CSVFolderConnector.type_name}', "
         f"'{HttpJsonConnector.type_name}', '{SFTPConnector.type_name}', "
         f"'{SQLConnector.type_name}', "
-        f"'{SalesforceConnector.type_name}'])"
+        f"'{SalesforceConnector.type_name}', "
+        f"'{StripeConnector.type_name}'])"
     )

infra/ingestion/stripe_connector.py

@@ -0,0 +1,222 @@
+"""
+infra.ingestion.stripe_connector — pull observations from a Stripe
+account via the REST API. Aimed at customer-health drift detection:
+who's delinquent, who has no default payment method, who's been
+around long enough that loss would hurt MRR.
+
+Customer flow:
+1. In Stripe Dashboard → Developers → API keys, mint a restricted
+   key with read access to Customers + Subscriptions.
+2. Store the key in an env var on this deployment (recommended)
+   OR paste inline into connector_config.
+3. Schedule a daily ingest with entity_type="customer".
+4. Engine baselines per-customer signals over time, drift triggers
+   on customers whose health metrics decay.
+
+Configuration:
+
+    api_key             (optional) the Stripe secret key, inline.
+                        Use ONLY for demos — secret ends up in the
+                        schedule's connector_config column.
+    api_key_env         (optional, recommended) env-var name whose
+                        value is the Stripe key. ONE of api_key /
+                        api_key_env required.
+    limit               max customers per fetch (default 100, max 100
+                        per Stripe page). v1: single page only.
+    timeout             HTTP timeout in seconds (default 30).
+    base_url            override for testing (default
+                        "https://api.stripe.com").
+
+v1 emits 3 metrics per customer, all derivable from /v1/customers
+without needing a second round-trip per row:
+
+    delinquent          1.0 if customer.delinquent else 0.0 — a
+                        Stripe-managed flag set when invoices fail.
+                        Rising → churn risk.
+    has_default_payment 1.0 if invoice_settings.default_payment_method
+                        is set else 0.0. Dropping → customer removed
+                        their card, often a churn precursor.
+    days_since_created  customer.created → days. Doesn't drift by
+                        itself, but the engine uses it as a baseline
+                        feature to weight other signals (old customer
+                        going delinquent is worse than a new one).
+
+v2 (future): aggregate /v1/subscriptions per customer for MRR /
+status / cancel_at_period_end. Out of v1 scope.
+
+Stdlib only — ``urllib.request``. All failures surface as
+:class:`ConnectorFetchError` matching the HTTP connector's shape.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+from datetime import datetime, timezone
+from typing import List, Optional
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+from core.pipeline import Observation
+
+from .connectors import Connector
+from .http_connector import ConnectorFetchError
+
+
+class StripeConnector(Connector):
+    """Fetches per-customer observations from a Stripe account."""
+
+    type_name = "stripe"
+
+    def __init__(
+        self, *,
+        api_key: Optional[str] = None,
+        api_key_env: Optional[str] = None,
+        limit: int = 100,
+        timeout: float = 30.0,
+        base_url: str = "https://api.stripe.com",
+    ):
+        if not (api_key or api_key_env):
+            raise ValueError(
+                "stripe connector needs api_key OR api_key_env "
+                "(recommended). Set one of them in connector_config."
+            )
+        if limit < 1 or limit > 100:
+            raise ValueError(
+                "stripe connector limit must be in [1, 100] — Stripe's "
+                "max page size for /v1/customers."
+            )
+        self.api_key = api_key
+        self.api_key_env = api_key_env
+        self.limit = limit
+        self.timeout = timeout
+        self.base_url = base_url.rstrip("/")
+
+    # ---- contract methods -----------------------------------------
+
+    def fetch(self, entity_type: str) -> List[Observation]:
+        if entity_type != "customer":
+            # v1 supports only "customer" — return empty rather than
+            # raise, so a schedule misconfigured for a different
+            # entity_type fails cleanly downstream (zero observations
+            # = pipeline notes nothing changed) instead of breaking.
+            return []
+        key = self._resolve_key()
+        url = f"{self.base_url}/v1/customers?limit={self.limit}"
+        req = Request(url, headers={
+            "Authorization": f"Bearer {key}",
+            "Accept": "application/json",
+        })
+        try:
+            with urlopen(req, timeout=self.timeout) as resp:
+                body = resp.read()
+        except HTTPError as e:
+            code, msg = _classify_http_error(e)
+            raise ConnectorFetchError(
+                code,
+                f"Stripe {e.code} {e.reason}: {msg}",
+                cause=e,
+            ) from e
+        except URLError as e:
+            raise ConnectorFetchError(
+                "network",
+                f"could not reach Stripe at {self.base_url}: {e.reason}",
+                cause=e,
+            ) from e
+        try:
+            payload = json.loads(body)
+        except json.JSONDecodeError as e:
+            raise ConnectorFetchError(
+                "bad_payload", "Stripe returned non-JSON body", cause=e,
+            ) from e
+        return self._customers_to_observations(payload)
+
+    def entity_types(self) -> List[str]:
+        return ["customer"]
+
+    # ---- internals ------------------------------------------------
+
+    def _resolve_key(self) -> str:
+        if self.api_key:
+            return self.api_key
+        key = os.environ.get(self.api_key_env or "")
+        if not key:
+            raise ConnectorFetchError(
+                "missing_secret",
+                f"env var {self.api_key_env!r} (api_key_env) is unset "
+                "or empty — set it on the deployment or pass api_key "
+                "inline in connector_config",
+            )
+        return key
+
+    def _customers_to_observations(self, payload: dict) -> List[Observation]:
+        if not isinstance(payload, dict) or "data" not in payload:
+            raise ConnectorFetchError(
+                "bad_payload",
+                "Stripe response missing 'data' — body keys: "
+                f"{list(payload.keys()) if isinstance(payload, dict) else type(payload).__name__}",
+            )
+        today_iso = datetime.now(timezone.utc).date().isoformat()
+        now_unix = time.time()
+        out: List[Observation] = []
+        for c in payload["data"]:
+            if not isinstance(c, dict):
+                continue
+            cid = c.get("id")
+            if not cid:
+                continue
+            created = c.get("created") or 0
+            days_since = max(0.0, (now_unix - float(created)) / 86400.0) if created else 0.0
+            default_pm = (
+                (c.get("invoice_settings") or {}).get("default_payment_method")
+            )
+            out.append(Observation(
+                entity_id=str(cid),
+                day=today_iso,
+                values={
+                    "delinquent":
+                        1.0 if c.get("delinquent") else 0.0,
+                    "has_default_payment":
+                        1.0 if default_pm else 0.0,
+                    "days_since_created": days_since,
+                },
+            ))
+        if payload.get("has_more"):
+            import sys as _sys
+            print(
+                f"warning: Stripe customers list has more than {self.limit} "
+                f"records (has_more=true). v1 reads a single page; tighten "
+                "the customer filter or wait for v2 pagination.",
+                file=_sys.stderr,
+            )
+        return out
+
+
+def _classify_http_error(e: HTTPError) -> tuple:
+    """Map a Stripe HTTP error to a stable (code, hint) tuple.
+
+    Stripe uses standard HTTP codes + an error object in the body
+    (``{"error": {"type": "...", "message": "..."}}``).
+    """
+    try:
+        body = e.read().decode("utf-8", errors="replace")
+    except Exception:
+        body = ""
+    snippet = body[:300] if body else ""
+    if e.code == 401:
+        return ("invalid_credentials",
+                f"Stripe rejected the API key — verify the key is "
+                f"correct and has read scope on Customers. Body: {snippet}")
+    if e.code == 403:
+        return ("forbidden",
+                f"Stripe key lacks scope for /v1/customers — restricted "
+                f"keys must include the customers:read permission: "
+                f"{snippet}")
+    if e.code == 429:
+        return ("rate_limit",
+                f"Stripe rate limit hit — schedule a less-frequent "
+                f"ingest or stagger calls. Body: {snippet}")
+    if 500 <= e.code < 600:
+        return ("upstream",
+                f"Stripe returned 5xx ({snippet})")
+    return ("http_error", snippet)