# Connectors and syncs

Source: /docs/reference/python/connectors-and-syncs.md

A connector owns source access and dataset discovery. A sync reads one dataset through that connector and yields Pandas DataFrames. Use the [source guides](/docs/connectors) for data overviews and worked starting points, or [Custom API](/docs/connectors/custom-api) for the shared HTTP pattern.

Each connector resource defines one `Connector` subclass in `connector.py`, a connection JSON Schema in `config.json`, and Python dependencies in `requirements.in`. Each sync defines one `Sync` subclass and its own schema and dependency file. Helpers can live beside the entrypoint.

## Connector [#connector]

### Lifecycle and methods [#lifecycle-and-methods]

```text
Connector(config: dict)
client(self) -> source-specific client
check(self) -> bool
discover(self) -> list[DiscoveredTable]
close(self) -> None
```

The runtime supplies resolved configuration. The inherited initializer stores
`self.config` and sets `self._client = None`. Preserve that initialization if
overriding it. All methods are synchronous.

- `client`: create a client lazily, cache it on self._client and reuse it. No
  network access during module import.
- `check`: perform a lightweight authenticated read verifying required access.
  Return exactly True on success; raise an informative exception on failure.
- `discover`: return stable unique dataset names and source column metadata. The
  worker checks connectivity before discovery. Handle catalog pagination.
- `close`: override if you own resources. Release them after success or failure;
  make repeated cleanup safe, and never create a client merely to close it.
  Cleanup does not commit source changes or advance ingestion checkpoints.

Connector implements a context manager: entering returns itself without connecting;
exiting calls close and propagates errors. Optional `vendor_domain: str | None`
is optional display/branding metadata (for example "stripe.com"), not a
classification or execution hook. Custom/internal connectors can leave it unset. Configuration, spec, compute,
mode and schedule are not connector class metadata.

### Configuration [#configuration]

`config.json` is a JSON Schema describing connection fields, not actual user values.
Use descriptions, defaults and field order where useful. Mark credentials with
`x-vortexify-secret: true`. For example:

```json
{
  "type": "object",
  "properties": {
    "base_url": {"type": "string", "format": "uri", "order": 0},
    "token": {"type": "string", "x-vortexify-secret": true, "order": 1}
  },
  "required": ["base_url", "token"],
  "additionalProperties": false
}
```

The platform stores protected secret references, resolves them in the worker and
validates resolved configuration. Do not interpret reference objects as plaintext
credentials. Keep real values out of source, schema defaults, tests and logs.
After saving the schema, supply values through the configuration service. Invalid or missing required values fail validation before extraction.

### Discovery contract [#discovery-contract]

```text
Column(name: str, type: str, nullable: bool = True)
DiscoveredTable(name: str, schema: list[Column], preview: list[dict] | None = None)
```

Names/type strings must be nonempty. Column names within each table are unique;
table names in discovery results must be unique and stable for later Sync planning.
Use `schema=` when constructing a DiscoveredTable; `.schema` reads columns and the
internal Pydantic field is schema_. Types describe source types, not guessed values
from one sample. Unexpected extra model fields are rejected.

Preview is optional, JSON-compatible, and limited to 100 rows per table. The SDK
does not fetch it automatically. Use an explicit bounded source read or a separate
Sync.preview instance with a fresh connector and temporary empty state. Do not
extract full tables just to discover them. An empty catalog can be valid; distinguish
it from missing permission, a failed request or malformed metadata. Discovery alone
does not create platform ingestion tables, Sync definitions or scheduled runs.

### Custom API example [#custom-api-example]

This example assumes the vendor documents GET /account for authentication and
GET /catalog returning `tables` with `name` and `columns`. Replace those assumptions
with the real source contract. The example needs `vortexify` and `requests` in `requirements.in`. Its endpoint and response shapes are illustrative; replace them with the real vendor contract.

```python
import requests
from vortexify import Column, Connector, DiscoveredTable

class CatalogConnector(Connector):
    vendor_domain = None

    def client(self):
        if self._client is None:
            self._client = requests.Session()
            self._client.headers["Authorization"] = "Bearer " + self.config["token"]
        return self._client

    def _get(self, path):
        response = self.client().get(
            self.config["base_url"].rstrip("/") + path, timeout=30,
        )
        response.raise_for_status()
        return response.json()

    def check(self):
        self._get("/account")
        return True

    def discover(self):
        return [
            DiscoveredTable(
                name=table["name"],
                schema=[
                    Column(name=column["name"], type=column["type"],
                           nullable=column.get("nullable", True))
                    for column in table["columns"]
                ],
            )
            for table in self._get("/catalog")["tables"]
        ]

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

For paginated vendors, extend discover to read the full catalog. Use suitable
request timeouts, documented rate limits, and bounded retries only for safe reads.
Sanitize errors when vendor URLs, headers or response bodies may expose secrets.

### Check access and discover datasets [#check-access-and-discover-datasets]

Checks and discovery execute saved source with the configured account. They are live source operations, separate from transform/API mock tests. Inspect their results before claiming connectivity or schema coverage. A source save or dependency build does not prove access, and discovery alone does not create syncs or publish their output tables.

## Sync [#sync]

The default workspace path is `connectors/<connector>/syncs/<sync>/`, beneath the project's configured connectors path. Use `connector/sync` in CLI selection and `Table("connector.sync")` in Python inputs. Connector and sync source commit independently; connector builds exclude nested sync code.

### Construction and class declarations [#construction-and-class-declarations]

```text
Sync(connector: Connector, config: dict | None = None, state: dict | None = None)
mode: "full" | "incremental" = "full"
compute: TransformCompute | None = None
schedule: CronSchedule | None = None
extract(self, *, avg_row_size_bytes: float | None = None) -> Iterator[pandas.DataFrame]
preview(self) -> pandas.DataFrame
```

The base initializer stores self.connector, sets self.client to connector.client(),
and stores supplied dictionaries as self.config and self.state (empty dictionaries
when omitted). In a Sync, self.client is the source client object, not a method.
Preserve this initialization if overriding it. Do not connect at module import.

`mode="full"` declares full extraction; `incremental` uses explicit author-defined
checkpoint logic. The SDK does not invent cursors or increment state automatically.
`compute=TransformCompute(size="small")` requests resources. Fields memory/disk are
MiB and cores is CPU allocation; unspecified compute/fields may be None, so handle
fallbacks when sizing batches. Supported size presets are xsmall/small/medium/large/
xlarge. The small preset is memory=2048, cores=1, disk=4096; explicit values override
presets. These are requested allocations, not measured free memory.

`schedule=CronSchedule(expression="0 * * * *")` enables cron scheduling. Default
None means no automatic schedule. A Sync supports no TriggerSchedule, schedule
lists, any/all trigger policy, or separate test/iter_batches/preview_rows SDK hooks.
Configuration and spec class attributes are not configuration metadata: use the
root JSON Schema. Credentials belong to platform-managed configuration.

### Extraction and checkpoint ownership [#extraction-and-checkpoint-ownership]

Implement synchronous `extract` as a generator yielding pandas DataFrames. Accept
the keyword avg_row_size_bytes even if you do not use it. It is an optional saved
estimate of average UTF-8 NDJSON row size, not a checkpoint. None means unknown.
Plan bounded batches; do not assume all source rows fit memory. Preserve column
names and source types according to the assigned output contract.

Read/write self.state deliberately for incremental extraction. Checkpoint semantics
must agree with when the worker successfully accepts output; do not move a cursor
past failed or unprocessed pages. Avoid source-side writes or cleanup that advances
checkpoints. The worker owns durable execution and publishing; yielding a batch is
not proof that a complete run has been committed.

### Preview contract [#preview-contract]

`preview()` is optional; the default raises NotImplementedError. Implement an
explicit bounded source read returning a small DataFrame. There is no fallback to
extract(). The caller must use a separate Sync instance with empty temporary state,
copied config and a fresh connector/client, and must never persist preview state.
Do not use an active extraction client or advance a real checkpoint for sampling.

`SyncPreview.from_dataframe(data, *, limit=100)` copies up to 1..100 rows, resets
the index and returns SyncPreview(data=<DataFrame>, avg_row_size_bytes=<float|None>).
It calculates UTF-8 NDJSON bytes including newlines using ISO dates; an empty sample
has no estimate. It makes no source requests, does not access Sync state and does
not call preview automatically. avg_row_size_bytes is only a batching hint.

### Example [#example]

This example assumes the assigned connector returns a client with documented
list_orders(page, page_size) yielding objects with a rows list and next_page token.
Adapt pagination and schemas to the actual vendor; do not invent those methods.

```python
import pandas as pd
from vortexify import CronSchedule, Sync, TransformCompute

class OrdersSync(Sync):
    mode = "full"
    compute = TransformCompute(size="small")
    schedule = CronSchedule(expression="0 * * * *")

    def extract(self, *, avg_row_size_bytes=None):
        page = None
        while True:
            response = self.client.list_orders(page=page, page_size=self.config["page_size"])
            rows = response["rows"]
            if rows:
                yield pd.DataFrame.from_records(rows)
            page = response.get("next_page")
            if page is None:
                break

    def preview(self):
        response = self.client.list_orders(page=None, page_size=10)
        return pd.DataFrame.from_records(response["rows"])
```

The corresponding root config.json contains a schema, not user values:

```json
{
  "type": "object",
  "properties": {"page_size": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 100}},
  "required": ["page_size"],
  "additionalProperties": false
}
```

No checkpoint is needed for this full extraction example. For incremental sources,
follow the documented cursor semantics and required replay/idempotency strategy.
Inspect preview/run results and logs to verify pagination, permissions, and output. SDK mock-test decorators apply to transform/API functions rather than Sync classes.

## From extraction to a usable table [#from-extraction-to-a-usable-table]

A sync execution publishes its output inventory and waits for schema discovery. Separate knowledge graph work then checks keys and relationships. Successful ingestion is enough to inspect known columns; wait for the relevant analysis when planning inferred joins. See [data readiness](/docs/building/bring-in-data#wait-for-the-result-your-next-step-needs).

Preview sync execution uses committed source/configuration and runs extraction. Checkpoints and accumulated inventories are scoped to the producer version, rather than a separate preview/run-mode checkpoint namespace. Use the independent `Sync.preview()` method for disposable sampling; a full preview run can advance that version's checkpoint. Production sync execution uses the deployed connector definition. Configuration changes save versions; deploying the updated version determines when production uses them.

In the platform, connector Tables lists the outputs of its syncs. Open a sync for Runs & logs, or a table for Data, Schema, and Relationships. The current resource workspace does not expose the connector check/discovery submission controls; those services remain distinct from sync ingestion.

## Register, configure, and run [#register-configure-and-run]

```sh
vortexify create connector shop
vortexify create sync shop/orders
vortexify commit --type connector --resource shop
vortexify connector shop --configure
vortexify commit --type sync --resource shop/orders
vortexify run sync shop/orders --wait
```

Edit the created files before committing. Connection and sync schemas describe fields, while actual values are supplied through platform configuration. Connection values can be edited with the interactive CLI command. Sync values use the separate `data/syncs/configure` service; the current mounted frontend has no dedicated sync configuration form. Preview uses a fresh connector and disposable state; production checkpoints belong to extraction.

See [compute and scheduling](/docs/reference/python/compute-and-schedules), [configuration](/docs/reference/cli/configuration), and [Bring in data](/docs/building/bring-in-data).