Stripe connector

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See Custom API for the shared SDK pattern and document-upload workflow.

What this connector is forCopy Anchor Link

Stripe is a payments and billing platform. A Vortexify Stripe connector provides access to the account's customer records, payment activity, recurring billing, product catalog, and movement of funds through the Stripe balance.

These datasets can support customer analysis, subscription reporting, payment/refund monitoring, and reconciliation. The connector owns the account connection and credentials; separate syncs turn particular Stripe resources into tables using the Python SDK.

Data and tables you can pullCopy Anchor Link

Stripe exposes API resources rather than SQL tables. The table names below are suggested datasets we can build from those resources. They describe the broader connector possibilities; the worked implementation later in this guide includes only customers.

This is a representative selection, not an exhaustive Stripe resource catalog. Actual availability depends on the account's Stripe products, permissions, environment, and API version. A table may legitimately be empty if the account does not use the corresponding product.

How the tables fit togetherCopy Anchor Link

Customer references connect customer records to subscriptions, invoices, and payment activity. Prices reference products; subscription items describe what a customer subscribes to. Invoice lines belong to a particular invoice and can be stored in a child table with the parent invoice ID included explicitly.

Payment intents, charges, refunds, payouts, and balance transactions describe different parts of the payment and settlement process. Choose the appropriate row meaning for your analysis rather than adding amounts from different tables as if they represented separate sales.

Keep object IDs and relationship references when shaping tables. Nested objects, arrays, and metadata can be preserved as JSON or normalized into child tables, depending on how the dataset will be used.

Extraction details to account forCopy Anchor Link

Resource-specific reads need more than an endpoint change. For example, invoice line items require iterating invoices and paginating each invoice's lines; the small embedded list on an invoice is not necessarily complete. The subscription list excludes canceled subscriptions by default, so decide which statuses the table should contain.

For each sync, define its columns, filters, pagination, and update strategy. A creation-time filter alone does not detect later changes to an older object. Start with a full read where appropriate and design incremental behavior separately.

Connecting to StripeCopy Anchor Link

Use a Stripe secret or restricted API key with read permissions for the resources you select. Keep test and live datasets separate. The example below uses the account's configured API version; validate response fields for that version when implementing each table. See authentication and API versioning.

Connection configuration belongs to the connector. Dataset-specific settings, such as filters or parent-resource selection, belong to each sync. Creating another dataset should reuse the connection while giving that sync its own extraction logic.

Worked example: sync customersCopy Anchor Link

The following example shows one full-read sync end to end: connection configuration, a reusable HTTP client, discovery of the implemented table, an independent preview, and paginated extraction. It demonstrates how to implement one dataset within the broader Stripe connector.

Example scopeCopy Anchor Link

This full read is not a transactionally consistent snapshot of a changing account. No live credentials or platform ingestion were used to validate this guide.

Example files and dependenciesCopy Anchor Link

The worked example requires permission to read customers. Its check() and discover() methods cover that resource; extend discovery alongside each additional sync you implement.

requirements.inCopy Anchor Link

The Python SDK must be available in the execution environment. Add these dependencies to the connector's requirements.in:

vortexify==0.1.1
pandas>=2,<3
httpx>=0.27,<1

The current Vortexify SDK is required in addition to HTTPX and Pandas. No separate native database driver is required. These dependency ranges are not a tested lockfile; resolve compatible versions for the runtime before deploying.

Register resources with the CLI workflow. With default project paths, the files below belong under:

connectors/stripe/
  connector.py
  config.json
  requirements.in
  syncs/customers/
    sync.py
    config.json

Connection configurationCopy Anchor Link

Save this JSON Schema as connectors/stripe/config.json. It describes configuration; it does not contain an actual API key.

{
  "type": "object",
  "properties": {
    "api_key": {
      "type": "string",
      "title": "API Key",
      "minLength": 1,
      "x-vortexify-secret": true,
      "order": 0
    }
  },
  "required": ["api_key"]
}

After committing the schema, use vortexify connector stripe --configure to provide the key. The platform supplies its value through self.config["api_key"].

Connector implementationCopy Anchor Link

Save this complete source file as connectors/stripe/connector.py.

import httpx

from vortexify import Column, Connector, DiscoveredTable


class StripeConnector(Connector):
    vendor_domain = "stripe.com"

    def client(self) -> httpx.Client:
        if self._client is None or self._client.is_closed:
            self._client = httpx.Client(
                base_url="https://api.stripe.com/v1/",
                headers={
                    "Authorization": f"Bearer {self.config['api_key']}",
                },
                timeout=60,
                follow_redirects=False,
            )
        return self._client

    def check(self) -> bool:
        response = self.client().get("customers", params={"limit": 1})
        response.raise_for_status()
        return True

    def discover(self) -> list[DiscoveredTable]:
        return [
            DiscoveredTable(
                name="customers",
                schema=[
                    Column(name="id", type="string", nullable=False),
                    Column(name="email", type="string"),
                    Column(name="name", type="string"),
                    Column(name="created", type="integer"),
                    Column(name="document", type="string"),
                ],
            ),
        ]

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.close()
            finally:
                self._client = None

The connector reuses one HTTP client. Its check performs a lightweight customer read, while discovery returns the schema of the implemented dataset without fetching samples. Exiting a connector context calls close() even when a request fails.

Sync configurationCopy Anchor Link

This sync has no dataset-specific settings. Save this schema as connectors/stripe/syncs/customers/config.json:

{
  "type": "object",
  "properties": {},
  "required": []
}

Sync implementationCopy Anchor Link

Save this complete source file as connectors/stripe/syncs/customers/sync.py.

import json
from collections.abc import Iterator
from math import isfinite

import pandas as pd

from vortexify import Sync


class StripeCustomersSync(Sync):
    mode = "full"
    MAX_PAGE_SIZE = 100
    MAX_BATCH_BYTES = 16 * 1024 * 1024
    COLUMNS = ["id", "email", "name", "created", "document"]

    def _read_page(
        self,
        *,
        limit: int,
        starting_after: str | None = None,
    ) -> tuple[list[dict], bool]:
        params = {"limit": limit}
        if starting_after is not None:
            params["starting_after"] = starting_after

        response = self.client.get("customers", params=params)
        response.raise_for_status()
        payload = response.json()
        records = payload["data"]
        has_more = payload["has_more"]

        if not isinstance(records, list) or any(
            not isinstance(record, dict) for record in records
        ):
            raise ValueError("Stripe returned an unexpected customer list")
        if not isinstance(has_more, bool):
            raise ValueError("Stripe returned an invalid has_more value")
        if has_more and not records:
            raise ValueError("Stripe returned has_more without customers")
        return records, has_more

    def _to_frame(self, records: list[dict]) -> pd.DataFrame:
        rows = [
            {
                "id": record["id"],
                "email": record.get("email"),
                "name": record.get("name"),
                "created": record.get("created"),
                "document": json.dumps(record, ensure_ascii=False),
            }
            for record in records
        ]
        return pd.DataFrame(rows, columns=self.COLUMNS, dtype=object)

    def _batch_size(self, avg_row_size_bytes: float | None) -> int:
        if avg_row_size_bytes is None:
            return self.MAX_PAGE_SIZE
        if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
            raise ValueError("avg_row_size_bytes must be finite and positive")

        memory_mib = self.compute.memory if self.compute else None
        if memory_mib is None or memory_mib <= 0:
            memory_mib = 1024
        target_bytes = min(
            self.MAX_BATCH_BYTES,
            max(1, memory_mib * 1024 * 1024 // 20),
        )
        # Reserve headroom for Python objects and serialization.
        estimated_rows = int(target_bytes / max(1.0, avg_row_size_bytes) / 4)
        return min(self.MAX_PAGE_SIZE, max(1, estimated_rows))

    def preview(self) -> pd.DataFrame:
        records, _ = self._read_page(limit=self.MAX_PAGE_SIZE)
        return self._to_frame(records[:self.MAX_PAGE_SIZE])

    def extract(
        self,
        *,
        avg_row_size_bytes: float | None = None,
    ) -> Iterator[pd.DataFrame]:
        batch_size = self._batch_size(avg_row_size_bytes)
        starting_after = None
        seen_cursors = set()

        while True:
            records, has_more = self._read_page(
                limit=batch_size,
                starting_after=starting_after,
            )
            if has_more:
                next_cursor = records[-1]["id"]
                if not isinstance(next_cursor, str) or not next_cursor:
                    raise ValueError("Stripe returned an invalid customer cursor")
                if next_cursor in seen_cursors:
                    raise ValueError("Stripe repeated a pagination cursor")
                seen_cursors.add(next_cursor)

            for start in range(0, len(records), batch_size):
                yield self._to_frame(records[start:start + batch_size])

            if not has_more:
                return
            starting_after = next_cursor

Preview and extraction share request/row-conversion helpers, but preview never invokes extraction. Cursor variables are local to extraction; this full-read sync does not use self.state. The backend must still construct a fresh connector and sync with disposable state for preview, as described in the preview lifecycle.

The batch-size formula is an example policy, not a guarantee enforced by the SDK. It reserves memory headroom and caps request/output batches at 100 rows. The size estimate may not represent unusually large customer records; a one-row minimum cannot bound the size of a single response record.

document preserves each complete customer as a JSON string. Other columns are selected scalar fields, with created kept as the source's integer timestamp rather than converted to a Pandas datetime.

Failures and limitationsCopy Anchor Link

HTTP failures, including authentication and rate-limit responses, propagate to the caller. This example does not implement request retries. It also rejects malformed pagination and repeated continuation cursors rather than silently claiming a complete read. Earlier batches may have been yielded before a later request fails; the platform must decide publication/retry behavior.

This implementation does not provide incremental extraction, deletion capture, or streaming. Use the table overview above to choose additional resources, then define a separate schema, pagination strategy, and preview for each sync.

Sources and further readingCopy Anchor Link

• Stripe API reference — The wider resource catalog and Stripe API conventions. The table overview above links to individual resource references.
• List customers — The endpoint used by this sync, its filters, and its response format.
• Customer object — Customer fields, their types, and nullable values.
• Pagination — Page-size limits, continuation cursors, and has_more.
• Authentication — API-key authentication and restricted-key permissions.
• API versioning — How the account default and Stripe-Version header select response behavior.
• HTTPX clients — Reusing connections, configuring requests, and closing clients.
• Vortexify connectors and syncs — Client lifecycle, discovery, independent previews, and extraction contracts.
• Vortexify CLI workflow — Registering the connector and sync, committing source, and configuring credentials.