Connectors and syncs

A connector owns source access and dataset discovery. A sync reads one dataset through that connector and yields Pandas DataFrames. Use the source guides for data overviews and worked starting points, or 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.

ConnectorCopy Anchor Link

Lifecycle and methodsCopy Anchor Link

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.

ConfigurationCopy Anchor Link

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:

{
  "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 contractCopy Anchor Link

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 exampleCopy Anchor Link

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.

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 datasetsCopy Anchor Link

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.

SyncCopy Anchor Link

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 declarationsCopy Anchor Link

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 ownershipCopy Anchor Link

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 contractCopy Anchor Link

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=, 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.

ExampleCopy Anchor Link

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.

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:

{
  "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 tableCopy Anchor Link

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.

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 runCopy Anchor Link

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, configuration, and Bring in data.