# Stripe connector

Source: /docs/connectors/stripe.md

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow.

## What this connector is for [#what-this-connector-is-for]

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](/docs/reference/python/connectors-and-syncs).

## Data and tables you can pull [#data-and-tables-you-can-pull]

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`.

| Data area | Suggested table | One row represents | Data you can use | API reference |
| --- | --- | --- | --- | --- |
| Customers | `customers` | A customer | Identity, contact details, creation time, and metadata | [Customers](https://docs.stripe.com/api/customers) |
| Payments | `payment_intents` | A payment lifecycle | Payment amount, currency, status, and customer references | [Payment Intents](https://docs.stripe.com/api/payment_intents) |
| Charges | `charges` | A charge | Charged amounts, payment outcomes, and links to payment activity | [Charges](https://docs.stripe.com/api/charges) |
| Refunds | `refunds` | A refund | Refunded amounts, status, and the associated payment | [Refunds](https://docs.stripe.com/api/refunds) |
| Disputes | `disputes` | A disputed charge | Dispute reason, status, and related charge | [Disputes](https://docs.stripe.com/api/disputes) |
| Product catalog | `products` | A product or service | Product names, descriptions, active state, and metadata | [Products](https://docs.stripe.com/api/products) |
| Pricing | `prices` | A price definition | Product reference, currency, pricing, and recurring-billing settings | [Prices](https://docs.stripe.com/api/prices) |
| Recurring billing | `subscriptions` | A subscription | Customer relationships, status, cancellation, and billing items | [Subscriptions](https://docs.stripe.com/api/subscriptions) |
| Invoicing | `invoices` | An invoice | Customer billing, amounts due or paid, and invoice status | [Invoices](https://docs.stripe.com/api/invoices) |
| Invoice detail | `invoice_line_items` | A line within an invoice | Billed quantities, line amounts, and pricing references | [Invoice line items](https://docs.stripe.com/api/invoice-line-item) |
| Settlement | `payouts` | A payout | Payout amounts, currency, status, and timing | [Payouts](https://docs.stripe.com/api/payouts) |
| Balance activity | `balance_transactions` | A movement in the Stripe balance | Amounts, fees, net balance impact, and source references | [Balance transactions](https://docs.stripe.com/api/balance_transactions) |

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 together [#how-the-tables-fit-together]

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 for [#extraction-details-to-account-for]

Resource-specific reads need more than an endpoint change. For example, [invoice line items](https://docs.stripe.com/api/invoice-line-item/retrieve) require iterating invoices and paginating each invoice's lines; the small embedded list on an invoice is not necessarily complete. The [subscription list](https://docs.stripe.com/api/subscriptions/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 Stripe [#connecting-to-stripe]

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](https://docs.stripe.com/api/authentication) and [API versioning](https://docs.stripe.com/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 customers [#worked-example-sync-customers]

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 scope [#example-scope]


| Capability | This example |
| --- | --- |
| Dataset | Customers |
| Output columns | `id`, `email`, `name`, `created`, `document` |
| Read behavior | Full read with `starting_after` pagination |
| Preview | Independent read of up to 100 customers |
| Incremental checkpoints | Not implemented |
| Writes to Stripe | None |

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 dependencies [#example-files-and-dependencies]


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.in [#requirementsin]

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

```text
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](/docs/reference/cli/overview). With default project paths, the files below belong under:

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

### Connection configuration [#connection-configuration]

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

```json
{
  "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 implementation [#connector-implementation]

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

```python
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 configuration [#sync-configuration]

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

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

### Sync implementation [#sync-implementation]

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

```python
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](/docs/reference/python/connectors-and-syncs).

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 limitations [#failures-and-limitations]

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 reading [#sources-and-further-reading]

- [Stripe API reference](https://docs.stripe.com/api) — The wider resource catalog and Stripe API conventions. The table overview above links to individual resource references.
- [List customers](https://docs.stripe.com/api/customers/list) — The endpoint used by this sync, its filters, and its response format.
- [Customer object](https://docs.stripe.com/api/customers/object) — Customer fields, their types, and nullable values.
- [Pagination](https://docs.stripe.com/api/pagination) — Page-size limits, continuation cursors, and `has_more`.
- [Authentication](https://docs.stripe.com/api/authentication) — API-key authentication and restricted-key permissions.
- [API versioning](https://docs.stripe.com/api/versioning) — How the account default and `Stripe-Version` header select response behavior.
- [HTTPX clients](https://www.python-httpx.org/advanced/clients/) — Reusing connections, configuring requests, and closing clients.
- [Vortexify connectors and syncs](/docs/reference/python/connectors-and-syncs) — Client lifecycle, discovery, independent previews, and extraction contracts.
- [Vortexify CLI workflow](/docs/reference/cli/overview) — Registering the connector and sync, committing source, and configuring credentials.