Sources and connectors

A source is a named place that data comes from. A connector is the code that knows how to talk to that kind of place. Separating the two means you can add a new source with a few lines of configuration, and the core of Synchronoid never changes.

Sources

A Source has a name, a connector type and the settings that connector needs. It also carries the rate limit and retry policy for that provider.

import synchronoid as sy

registry = sy.Source(
    name="company_registry",
    connector="http_json",
    base_url="https://registry.example.org/api",
    auth=sy.Secret("REGISTRY_TOKEN"),
    rate_limit=sy.RateLimit(per_minute=120),
    retry=sy.Retry(max_attempts=5, backoff="exponential"),
)

You can also declare sources in synchronoid.yaml. This is useful when non-developers add sources, or when you generate them from a list.

sources:
  company_registry:
    connector: http_json
    base_url: https://registry.example.org/api
    rate_limit: { per_minute: 120 }

One connector, many sources

A connector describes a kind of system, such as "a JSON API with page-based pagination". A source is one concrete instance. You can create as many sources from one connector as you need, for example one per provider account, region or customer.

REGIONS = {"north": 60, "south": 60, "west": 30}

listing_sources = [
    sy.Source(
        name=f"listings_{region}",
        connector="http_json",
        base_url=f"https://{region}.listings.example.com/v2",
        rate_limit=sy.RateLimit(per_minute=limit),
    )
    for region, limit in REGIONS.items()
]

Each source has its own budget, health status and pause switch. You can pause listings_west without touching the others.

Built-in connectors

The full catalog is on the integrations page.

Writing a custom connector

If no connector fits, write your own with the connector SDK. A connector is a class that tells Synchronoid how to list what is available and how to fetch one unit of it. It does not deal with retries, rate limits or storage. The platform does that around your code.

import synchronoid as sy
from synchronoid.connectors import Connector, Page

@sy.connector(name="ledger_api")
class LedgerApi(Connector):
    # Settings a source must provide
    config = {"base_url": str, "account_id": str}

    def fetch(self, request, cursor=None) -> Page:
        resp = self.http.get(
            f"{self.config['base_url']}/accounts/{self.config['account_id']}/entries",
            params={"after": cursor, "limit": 500},
        )
        body = resp.json()
        return Page(records=body["entries"], next_cursor=body.get("next"))

    def check(self) -> bool:
        return self.http.get(f"{self.config['base_url']}/health").ok

Then use it like any built-in connector:

acme_ledger = sy.Source(
    name="acme_ledger",
    connector="ledger_api",
    base_url="https://ledger.example.com",
    account_id="acct_1042",
    rate_limit=sy.RateLimit(per_minute=90),
)

What the platform handles for you

• Rate limiting. self.http waits for budget from the source's shared rate limit before each request. Many pipelines can use one source without exceeding the quota.
• Retries and pausing. Errors from fetch follow the source's retry policy. A source that keeps failing is paused automatically and an operator is alerted.
• Raw storage. Every page you return is stored untouched in the raw layer, with provenance attached.

Secrets

Never put tokens in code. Use sy.Secret("NAME") and set the value in the workspace. See Secrets.