`graflo.architecture.contract.bindings.connectors`¶

Resource connector types (file, SQL table, SPARQL, API, Kafka).

Attributes¶

`TEMPLATE_VARIABLE = re.compile('\\{([A-Za-z_][A-Za-z0-9_]*)\\}')` `module-attribute` ¶

`TemplateValue = str | int | float | bool` `module-attribute` ¶

Classes¶

`APIConnector` ¶

Bases: TemplatedConnector

Connector for REST API endpoints.

Declares the non-secret access pattern (path, method, pagination). Runtime base_url and credentials are supplied via connector_connection -> conn_proxy -> :class:~graflo.connections.sources.RestApiConnConfig.

path and params are address templates and body a query text (see :class:TemplatedConnector); their variables come from values and from parent.fields.

Attributes:

Name	Type	Description
`path`	`str`	Relative endpoint path (e.g. `/api/users`).
`method`	`str`	HTTP method (default `GET`).
`params`	`dict[str, Any]`	Static query parameters.
`body`	`dict[str, Any] \| str \| None`	Request body: a mapping is sent as JSON, a string as is. Not allowed with `GET`.
`pagination`	`PaginationConfig \| None`	Pagination strategy and response path configuration.
`row_annotations`	`dict[str, Any]`	Constant fields merged into every fetched document (doc wins).
`headers`	`dict[str, str]`	Non-secret HTTP headers.
`timeout`	`float \| None`	Request timeout in seconds.
`retries`	`int`	Number of retry attempts. `0` fails on the first error.
`retry_backoff_factor`	`float`	Backoff factor for retries.
`retry_status_forcelist`	`list[int]`	HTTP status codes to retry on. A `Retry-After` header on the response sets the wait.
`verify`	`bool`	Verify SSL certificates.
`watermark`	`WatermarkConfig \| None`	Where a run left off; see :class:`WatermarkConfig`.
`parent`	`ParentRequest \| None`	Make one request per record of another API connector, with `{variable}` in `path`, `params` and `body` filled from that record; see :class:`ParentRequest`. Without its own `conn_proxy` mapping the child uses the parent's connection.
`values`	`dict[str, list[TemplateValue]] \| None`	One request per value of each variable; see :class:`TemplatedConnector`.

Source code in graflo/architecture/contract/bindings/connectors.py

class APIConnector(TemplatedConnector):
    """Connector for REST API endpoints.

    Declares the non-secret access pattern (path, method, pagination). Runtime
    ``base_url`` and credentials are supplied via ``connector_connection`` ->
    ``conn_proxy`` -> :class:`~graflo.connections.sources.RestApiConnConfig`.

    ``path`` and ``params`` are address templates and ``body`` a query text
    (see :class:`TemplatedConnector`); their variables come from ``values``
    and from ``parent.fields``.

    Attributes:
        path: Relative endpoint path (e.g. ``/api/users``).
        method: HTTP method (default ``GET``).
        params: Static query parameters.
        body: Request body: a mapping is sent as JSON, a string as is. Not
            allowed with ``GET``.
        pagination: Pagination strategy and response path configuration.
        row_annotations: Constant fields merged into every fetched document (doc wins).
        headers: Non-secret HTTP headers.
        timeout: Request timeout in seconds.
        retries: Number of retry attempts. ``0`` fails on the first error.
        retry_backoff_factor: Backoff factor for retries.
        retry_status_forcelist: HTTP status codes to retry on. A ``Retry-After``
            header on the response sets the wait.
        verify: Verify SSL certificates.
        watermark: Where a run left off; see :class:`WatermarkConfig`.
        parent: Make one request per record of another API connector, with
            ``{variable}`` in ``path``, ``params`` and ``body`` filled from that
            record; see :class:`ParentRequest`. Without its own ``conn_proxy``
            mapping the child uses the parent's connection.
        values: One request per value of each variable; see
            :class:`TemplatedConnector`.
    """

    path: str = Field(..., description="Relative API endpoint path")
    method: str = "GET"
    params: dict[str, Any] = Field(default_factory=dict)
    body: dict[str, Any] | str | None = None
    pagination: PaginationConfig | None = None
    headers: dict[str, str] = Field(default_factory=dict)
    timeout: float | None = None
    retries: int = 0
    retry_backoff_factor: float = 0.1
    retry_status_forcelist: list[int] = Field(
        default_factory=lambda: [429, 500, 502, 503, 504]
    )
    verify: bool = True
    watermark: WatermarkConfig | None = None
    parent: ParentRequest | None = None

    @model_validator(mode="after")
    def _check_body(self) -> Self:
        if self.body is not None and self.method.upper() == "GET":
            raise ValueError(
                f"API connector '{self._label()}' has a body; a GET request "
                "carries none"
            )
        return self

    def _address_templates(self) -> list[Any]:
        return [self.path, self.params]

    def _text_templates(self) -> list[Any]:
        return [self.body]

    def _bound_variables(self) -> set[str]:
        return set(self.parent.fields) if self.parent is not None else set()

    def _label(self) -> str:
        return self.name or self.path

    def rendered(self, values: Mapping[str, Any]) -> Self:
        return self.model_copy(
            update={
                "path": render_text(self.path, values, quote_values=True),
                "params": render_param(self.params, values),
                "body": None if self.body is None else render_body(self.body, values),
                "values": None,
            }
        )

    def _hash_payload(self) -> dict[str, Any]:
        payload = super()._hash_payload()
        for optional in ("body", "watermark", "parent"):
            if payload.get(optional) is None:
                # Unset, it must not move the hash of connectors declared before it.
                payload.pop(optional, None)
        return payload

    @staticmethod
    def _join_url(base_url: str, path: str) -> str:
        return f"{base_url.rstrip('/')}/{path.lstrip('/')}"

    def matches(self, resource_identifier: str) -> bool:
        """Match resource name, connector name, or path tail."""
        if self.name is not None and resource_identifier == self.name:
            return True
        if self.resource_name is not None and resource_identifier == self.resource_name:
            return True
        path_tail = self.path.rstrip("/").rsplit("/", 1)[-1]
        return resource_identifier in {self.path, path_tail}

    def bound_source_kind(self) -> BoundSourceKind:
        return BoundSourceKind.API

    def build_api_config(
        self,
        *,
        base_url: str,
        auth: ApiAuth | None = None,
        default_headers: dict[str, str] | None = None,
        page_size_override: int | None = None,
    ) -> APIConfig:
        """Merge contract fields with runtime connection config into ``APIConfig``."""
        from graflo.data_source.api import APIConfig

        headers = dict(default_headers or {})
        headers.update(self.headers)

        pagination = self.pagination
        if pagination is not None and page_size_override is not None:
            pagination = pagination.model_copy(
                update={
                    "request": pagination.request.model_copy(
                        update={"page_size": page_size_override}
                    )
                }
            )

        return APIConfig(
            url=self._join_url(base_url, self.path),
            method=self.method,
            headers=headers,
            auth=auth,
            params=dict(self.params),
            body=self.body,
            timeout=self.timeout,
            retries=self.retries,
            retry_backoff_factor=self.retry_backoff_factor,
            retry_status_forcelist=list(self.retry_status_forcelist),
            verify=self.verify,
            pagination=pagination,
            row_annotations=dict(self.row_annotations),
            watermark=self.watermark,
        )

Attributes¶

`body = None` `class-attribute` `instance-attribute` ¶

`headers = Field(default_factory=dict)` `class-attribute` `instance-attribute` ¶

`method = 'GET'` `class-attribute` `instance-attribute` ¶

`pagination = None` `class-attribute` `instance-attribute` ¶

`params = Field(default_factory=dict)` `class-attribute` `instance-attribute` ¶

`parent = None` `class-attribute` `instance-attribute` ¶

`path = Field(..., description='Relative API endpoint path')` `class-attribute` `instance-attribute` ¶

`retries = 0` `class-attribute` `instance-attribute` ¶

`retry_backoff_factor = 0.1` `class-attribute` `instance-attribute` ¶

`retry_status_forcelist = Field(default_factory=lambda: [429, 500, 502, 503, 504])` `class-attribute` `instance-attribute` ¶

`timeout = None` `class-attribute` `instance-attribute` ¶

`verify = True` `class-attribute` `instance-attribute` ¶

`watermark = None` `class-attribute` `instance-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    return BoundSourceKind.API

`build_api_config(*, base_url, auth=None, default_headers=None, page_size_override=None)` ¶

Merge contract fields with runtime connection config into APIConfig.

Source code in graflo/architecture/contract/bindings/connectors.py

def build_api_config(
    self,
    *,
    base_url: str,
    auth: ApiAuth | None = None,
    default_headers: dict[str, str] | None = None,
    page_size_override: int | None = None,
) -> APIConfig:
    """Merge contract fields with runtime connection config into ``APIConfig``."""
    from graflo.data_source.api import APIConfig

    headers = dict(default_headers or {})
    headers.update(self.headers)

    pagination = self.pagination
    if pagination is not None and page_size_override is not None:
        pagination = pagination.model_copy(
            update={
                "request": pagination.request.model_copy(
                    update={"page_size": page_size_override}
                )
            }
        )

    return APIConfig(
        url=self._join_url(base_url, self.path),
        method=self.method,
        headers=headers,
        auth=auth,
        params=dict(self.params),
        body=self.body,
        timeout=self.timeout,
        retries=self.retries,
        retry_backoff_factor=self.retry_backoff_factor,
        retry_status_forcelist=list(self.retry_status_forcelist),
        verify=self.verify,
        pagination=pagination,
        row_annotations=dict(self.row_annotations),
        watermark=self.watermark,
    )

`matches(resource_identifier)` ¶

Match resource name, connector name, or path tail.

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Match resource name, connector name, or path tail."""
    if self.name is not None and resource_identifier == self.name:
        return True
    if self.resource_name is not None and resource_identifier == self.resource_name:
        return True
    path_tail = self.path.rstrip("/").rsplit("/", 1)[-1]
    return resource_identifier in {self.path, path_tail}

`rendered(values)` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def rendered(self, values: Mapping[str, Any]) -> Self:
    return self.model_copy(
        update={
            "path": render_text(self.path, values, quote_values=True),
            "params": render_param(self.params, values),
            "body": None if self.body is None else render_body(self.body, values),
            "values": None,
        }
    )

`ApiResponseStructure` ¶

Bases: ConfigBaseModel

Maps JSON response envelope fields to extraction and pagination signals.

Source code in graflo/architecture/contract/bindings/connectors.py

class ApiResponseStructure(ConfigBaseModel):
    """Maps JSON response envelope fields to extraction and pagination signals."""

    records_path: str | None = Field(
        default=None,
        description="Dot path to the record list (e.g. ``results``).",
    )
    total_count_path: str | None = Field(
        default=None,
        description="Dot path to total item count across all pages (e.g. ``count``).",
    )
    offset_path: str | None = Field(
        default=None,
        description="Dot path to echoed page start index (e.g. ``offset``).",
    )
    next_offset_path: str | None = Field(
        default=None,
        description=(
            "Dot path to server-provided next offset for the following request "
            "(e.g. ``next_offset``)."
        ),
    )
    has_more_path: str | None = Field(
        default=None,
        description="Dot path to a boolean more-pages flag (e.g. ``has_more``).",
    )
    cursor_path: str | None = Field(
        default=None,
        description="Dot path to the next opaque cursor token.",
    )
    ok_path: str | None = Field(
        default=None,
        description=(
            "Dot path to a success flag in every response body (e.g. ``ok``). "
            "A body where it is not ``true`` raises, even with HTTP status 200."
        ),
    )
    error_path: str | None = Field(
        default=None,
        description=(
            "Dot path to an error code in the response body (e.g. ``error``). "
            "Named in the raised error; a non-empty value raises on its own."
        ),
    )
    batch_metadata_paths: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Map row annotation keys to response dot paths "
            "(e.g. ``_batch_id: result_id``)."
        ),
    )
    auto_detect: bool = Field(
        default=False,
        description=(
            "When true, infer unset response paths from the first response body."
        ),
    )

Attributes¶

`auto_detect = Field(default=False, description='When true, infer unset response paths from the first response body.')` `class-attribute` `instance-attribute` ¶

batch_metadata_paths = Field(default_factory=dict, description='Map row annotation keys to response dot paths (e.g. ``_batch_id: result_id``).') `class-attribute` `instance-attribute` ¶

`cursor_path = Field(default=None, description='Dot path to the next opaque cursor token.')` `class-attribute` `instance-attribute` ¶

error_path = Field(default=None, description='Dot path to an error code in the response body (e.g. ``error``). Named in the raised error; a non-empty value raises on its own.') `class-attribute` `instance-attribute` ¶

has_more_path = Field(default=None, description='Dot path to a boolean more-pages flag (e.g. ``has_more``).') `class-attribute` `instance-attribute` ¶

next_offset_path = Field(default=None, description='Dot path to server-provided next offset for the following request (e.g. ``next_offset``).') `class-attribute` `instance-attribute` ¶

offset_path = Field(default=None, description='Dot path to echoed page start index (e.g. ``offset``).') `class-attribute` `instance-attribute` ¶

ok_path = Field(default=None, description='Dot path to a success flag in every response body (e.g. ``ok``). A body where it is not ``true`` raises, even with HTTP status 200.') `class-attribute` `instance-attribute` ¶

records_path = Field(default=None, description='Dot path to the record list (e.g. ``results``).') `class-attribute` `instance-attribute` ¶

total_count_path = Field(default=None, description='Dot path to total item count across all pages (e.g. ``count``).') `class-attribute` `instance-attribute` ¶

`BoundSourceKind` ¶

Bases: BaseEnum

Physical source modality for a bound connector (how rows are retrieved).

This describes the connector-backed access pattern, not the abstract ingestion resource. File format (CSV, JSON, etc.) is chosen by the loader from file extensions.

Attributes:

Name	Type	Description
`FILE`		File-based connector (directory + pattern or paths).
`SQL_TABLE`		SQL table / database-backed connector.
`SPARQL`		SPARQL / RDF connector (endpoint or local RDF via rdflib).
`API`		REST API connector (path + pagination on a runtime base URL).
`KAFKA`		Kafka topic consumer (topics + group_id on a runtime bootstrap).
`PUSH`		Records handed to the resource by the caller; no address.

Source code in graflo/architecture/contract/bindings/connectors.py

class BoundSourceKind(BaseEnum):
    """Physical source modality for a bound connector (how rows are retrieved).

    This describes the connector-backed access pattern, not the abstract
    ingestion resource. File format (CSV, JSON, etc.) is chosen by the loader
    from file extensions.

    Attributes:
        FILE: File-based connector (directory + pattern or paths).
        SQL_TABLE: SQL table / database-backed connector.
        SPARQL: SPARQL / RDF connector (endpoint or local RDF via rdflib).
        API: REST API connector (path + pagination on a runtime base URL).
        KAFKA: Kafka topic consumer (topics + group_id on a runtime bootstrap).
        PUSH: Records handed to the resource by the caller; no address.
    """

    FILE = "file"
    SQL_TABLE = "sql_table"
    SPARQL = "sparql"
    API = "api"
    KAFKA = "kafka"
    PUSH = "push"

Attributes¶

`API = 'api'` `class-attribute` `instance-attribute` ¶

`FILE = 'file'` `class-attribute` `instance-attribute` ¶

`KAFKA = 'kafka'` `class-attribute` `instance-attribute` ¶

`PUSH = 'push'` `class-attribute` `instance-attribute` ¶

`SPARQL = 'sparql'` `class-attribute` `instance-attribute` ¶

`SQL_TABLE = 'sql_table'` `class-attribute` `instance-attribute` ¶

`ConnectorUpdate` ¶

Bases: ConfigBaseModel

Patch payload for an existing connector (applied after manifest load).

Only connector is a fixed field; any other keys are captured as extras and merged into the resolved connector via model_validate (so validators, including hash recomputation, run). New connector types and fields do not require changes to this model. Not part of the stored GraphManifest; load from a separate file or build in code, then call Bindings.apply_connector_update.

Attributes:

Name	Type	Description
`connector`	`str`	Connector `name` or `hash` (same resolution as bindings).

Source code in graflo/architecture/contract/bindings/connectors.py

class ConnectorUpdate(ConfigBaseModel):
    """Patch payload for an existing connector (applied after manifest load).

    Only ``connector`` is a fixed field; any other keys are captured as extras and
    merged into the resolved connector via ``model_validate`` (so validators,
    including hash recomputation, run). New connector types and fields do not
    require changes to this model. Not part of the stored ``GraphManifest``;
    load from a separate file or build in code, then call
    ``Bindings.apply_connector_update``.

    Attributes:
        connector: Connector ``name`` or ``hash`` (same resolution as bindings).
    """

    model_config = ConfigDict(extra="allow")

    connector: str = Field(
        ...,
        description="Connector reference: name or hash of the connector to patch.",
    )

    def as_patch(self) -> dict[str, Any]:
        """Return extra keys as a patch mapping (excludes ``connector``)."""
        return dict(self.model_extra or {})

Attributes¶

`connector = Field(..., description='Connector reference: name or hash of the connector to patch.')` `class-attribute` `instance-attribute` ¶

`model_config = ConfigDict(extra='allow')` `class-attribute` `instance-attribute` ¶

Methods:¶

`as_patch()` ¶

Return extra keys as a patch mapping (excludes connector).

Source code in graflo/architecture/contract/bindings/connectors.py

def as_patch(self) -> dict[str, Any]:
    """Return extra keys as a patch mapping (excludes ``connector``)."""
    return dict(self.model_extra or {})

`FileConnector` ¶

Bases: ResourceConnector

Connector for matching files.

Attributes:

Name	Type	Description
`regex`	`str \| None`	Regular expression pattern for matching filenames
`sub_path`	`Path`	Directory to search for matching files (default: "./"). With a `conn_proxy` naming a files connection it is relative to that connection's root, a local directory or an S3 prefix; without one, a local path.
`recursive`	`bool`	Also search the directories under `sub_path`. `regex` still matches the file's name, not its path.
`file_type`	`Literal['json', 'jsonl', 'table', 'parquet'] \| None`	Format of the matched files, for names that do not show it. Left unset, the format comes from the extension.
`time_filter`	`ColumnTimeFilter \| None`	Refused when set. A file is read whole, so a window declared here would not be applied; filter the rows in a pipeline step, or bind a :class:`TableConnector`.

Source code in graflo/architecture/contract/bindings/connectors.py

class FileConnector(ResourceConnector):
    """Connector for matching files.

    Attributes:
        regex: Regular expression pattern for matching filenames
        sub_path: Directory to search for matching files (default: "./"). With
            a ``conn_proxy`` naming a files connection it is relative to that
            connection's root, a local directory or an S3 prefix; without one,
            a local path.
        recursive: Also search the directories under ``sub_path``. ``regex``
            still matches the file's name, not its path.
        file_type: Format of the matched files, for names that do not show it.
            Left unset, the format comes from the extension.
        time_filter: Refused when set. A file is read whole, so a window
            declared here would not be applied; filter the rows in a pipeline
            step, or bind a :class:`TableConnector`.
    """

    takes_default_conn_proxy: ClassVar[bool] = False

    regex: str | None = None
    sub_path: pathlib.Path = Field(default_factory=lambda: pathlib.Path("./"))
    recursive: bool = False
    file_type: Literal["json", "jsonl", "table", "parquet"] | None = None
    time_filter: ColumnTimeFilter | None = None

    def _hash_payload(self) -> dict[str, Any]:
        payload = super()._hash_payload()
        # Left out at their defaults, so a connector that does not set them
        # keeps the hash it is referred to by.
        if not self.recursive:
            del payload["recursive"]
        if self.file_type is None:
            del payload["file_type"]
        return payload

    @model_validator(mode="after")
    def _validate_file_connector(self) -> Self:
        """Ensure sub_path is a Path."""
        if not isinstance(self.sub_path, pathlib.Path):
            object.__setattr__(self, "sub_path", pathlib.Path(self.sub_path))
        if self.row_annotations:
            raise ValueError("row_annotations is not implemented for FileConnector")
        if self.time_filter is not None:
            raise ValueError(
                "time_filter is not implemented for FileConnector: a file is read "
                "whole, so the window would not be applied"
            )
        return self

    def matches(self, resource_identifier: str) -> bool:
        """Check if connector matches a filename.

        Args:
            resource_identifier: Filename to match

        Returns:
            bool: True if connector matches
        """
        if self.regex is None:
            return False
        return bool(re.match(self.regex, resource_identifier))

    def bound_source_kind(self) -> BoundSourceKind:
        """File connector always uses ``BoundSourceKind.FILE``."""
        return BoundSourceKind.FILE

Attributes¶

`file_type = None` `class-attribute` `instance-attribute` ¶

`recursive = False` `class-attribute` `instance-attribute` ¶

`regex = None` `class-attribute` `instance-attribute` ¶

`sub_path = Field(default_factory=lambda: pathlib.Path('./'))` `class-attribute` `instance-attribute` ¶

`takes_default_conn_proxy = False` `class-attribute` ¶

`time_filter = None` `class-attribute` `instance-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

File connector always uses BoundSourceKind.FILE.

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    """File connector always uses ``BoundSourceKind.FILE``."""
    return BoundSourceKind.FILE

`matches(resource_identifier)` ¶

Check if connector matches a filename.

Parameters:

Name	Type	Description	Default
`resource_identifier`	`str`	Filename to match	required

Returns:

Name	Type	Description
`bool`	`bool`	True if connector matches

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Check if connector matches a filename.

    Args:
        resource_identifier: Filename to match

    Returns:
        bool: True if connector matches
    """
    if self.regex is None:
        return False
    return bool(re.match(self.regex, resource_identifier))

`KafkaConnector` ¶

Bases: ResourceConnector

Connector for Kafka topic consumption.

Declares the non-secret access pattern (topics, consumer group, decode options). Runtime bootstrap servers and credentials are supplied via connector_connection -> conn_proxy -> :class:~graflo.connections.sources.KafkaConnConfig.

Attributes:

Name	Type	Description
`topics`	`list[str]`	Topic names to subscribe to.
`group_id`	`str`	Kafka consumer group id.
`auto_offset_reset`	`Literal['earliest', 'latest']`	Where to start when no committed offset exists.
`value_encoding`	`Literal['json']`	Payload decode mode (`json` only in v1).
`include_headers`	`bool`	When True, attach decoded headers under `_kafka_headers`.
`idle_ms`	`int`	Stop after this many milliseconds with no messages.
`max_wait_ms`	`int \| None`	Hard wall-clock cap for a single `iter_batches` run.
`poll_timeout_ms`	`int`	Per-poll timeout passed to the Kafka client.
`row_annotations`	`dict[str, Any]`	Constant fields merged into every decoded document (doc wins).

Source code in graflo/architecture/contract/bindings/connectors.py

class KafkaConnector(ResourceConnector):
    """Connector for Kafka topic consumption.

    Declares the non-secret access pattern (topics, consumer group, decode
    options). Runtime bootstrap servers and credentials are supplied via
    ``connector_connection`` -> ``conn_proxy`` ->
    :class:`~graflo.connections.sources.KafkaConnConfig`.

    Attributes:
        topics: Topic names to subscribe to.
        group_id: Kafka consumer group id.
        auto_offset_reset: Where to start when no committed offset exists.
        value_encoding: Payload decode mode (``json`` only in v1).
        include_headers: When True, attach decoded headers under ``_kafka_headers``.
        idle_ms: Stop after this many milliseconds with no messages.
        max_wait_ms: Hard wall-clock cap for a single ``iter_batches`` run.
        poll_timeout_ms: Per-poll timeout passed to the Kafka client.
        row_annotations: Constant fields merged into every decoded document (doc wins).
    """

    topics: list[str] = Field(..., min_length=1, description="Kafka topics to consume")
    group_id: str = Field(..., description="Consumer group id")
    auto_offset_reset: Literal["earliest", "latest"] = "earliest"
    value_encoding: Literal["json"] = "json"
    include_headers: bool = False
    idle_ms: int = Field(
        default=2000,
        ge=0,
        description="Stop after this many ms with no messages (0 disables idle stop).",
    )
    max_wait_ms: int | None = Field(
        default=None,
        ge=0,
        description="Optional wall-clock cap for one consume run.",
    )
    poll_timeout_ms: int = Field(
        default=500,
        ge=1,
        description="Timeout for each consumer poll call.",
    )

    @field_validator("topics")
    @classmethod
    def _non_empty_topic_names(cls, v: list[str]) -> list[str]:
        cleaned = [t.strip() for t in v if t and t.strip()]
        if not cleaned:
            raise ValueError("topics must contain at least one non-empty topic name")
        return cleaned

    def matches(self, resource_identifier: str) -> bool:
        """Match resource name, connector name, or a subscribed topic."""
        if self.name is not None and resource_identifier == self.name:
            return True
        if self.resource_name is not None and resource_identifier == self.resource_name:
            return True
        return resource_identifier in self.topics

    def bound_source_kind(self) -> BoundSourceKind:
        return BoundSourceKind.KAFKA

    def build_kafka_config(
        self,
        *,
        conn: KafkaConnConfig,
    ) -> KafkaConfig:
        """Merge contract fields with runtime connection config into ``KafkaConfig``."""
        from graflo.data_source.kafka import KafkaConfig

        return KafkaConfig(
            bootstrap_servers=conn.bootstrap_servers,
            security_protocol=conn.security_protocol,
            client_id=conn.client_id,
            sasl_mechanism=conn.sasl_mechanism,
            sasl_username=conn.sasl_username,
            sasl_password=conn.sasl_password,
            topics=list(self.topics),
            group_id=self.group_id,
            auto_offset_reset=self.auto_offset_reset,
            value_encoding=self.value_encoding,
            include_headers=self.include_headers,
            idle_ms=self.idle_ms,
            max_wait_ms=self.max_wait_ms,
            poll_timeout_ms=self.poll_timeout_ms,
            row_annotations=dict(self.row_annotations),
        )

Attributes¶

`auto_offset_reset = 'earliest'` `class-attribute` `instance-attribute` ¶

`group_id = Field(..., description='Consumer group id')` `class-attribute` `instance-attribute` ¶

`idle_ms = Field(default=2000, ge=0, description='Stop after this many ms with no messages (0 disables idle stop).')` `class-attribute` `instance-attribute` ¶

`include_headers = False` `class-attribute` `instance-attribute` ¶

`max_wait_ms = Field(default=None, ge=0, description='Optional wall-clock cap for one consume run.')` `class-attribute` `instance-attribute` ¶

`poll_timeout_ms = Field(default=500, ge=1, description='Timeout for each consumer poll call.')` `class-attribute` `instance-attribute` ¶

`topics = Field(..., min_length=1, description='Kafka topics to consume')` `class-attribute` `instance-attribute` ¶

`value_encoding = 'json'` `class-attribute` `instance-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    return BoundSourceKind.KAFKA

`build_kafka_config(*, conn)` ¶

Merge contract fields with runtime connection config into KafkaConfig.

Source code in graflo/architecture/contract/bindings/connectors.py

def build_kafka_config(
    self,
    *,
    conn: KafkaConnConfig,
) -> KafkaConfig:
    """Merge contract fields with runtime connection config into ``KafkaConfig``."""
    from graflo.data_source.kafka import KafkaConfig

    return KafkaConfig(
        bootstrap_servers=conn.bootstrap_servers,
        security_protocol=conn.security_protocol,
        client_id=conn.client_id,
        sasl_mechanism=conn.sasl_mechanism,
        sasl_username=conn.sasl_username,
        sasl_password=conn.sasl_password,
        topics=list(self.topics),
        group_id=self.group_id,
        auto_offset_reset=self.auto_offset_reset,
        value_encoding=self.value_encoding,
        include_headers=self.include_headers,
        idle_ms=self.idle_ms,
        max_wait_ms=self.max_wait_ms,
        poll_timeout_ms=self.poll_timeout_ms,
        row_annotations=dict(self.row_annotations),
    )

`matches(resource_identifier)` ¶

Match resource name, connector name, or a subscribed topic.

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Match resource name, connector name, or a subscribed topic."""
    if self.name is not None and resource_identifier == self.name:
        return True
    if self.resource_name is not None and resource_identifier == self.resource_name:
        return True
    return resource_identifier in self.topics

`PaginationConfig` ¶

Bases: ConfigBaseModel

Configuration for API pagination (contract-level, secret-free).

Combines request construction (request) with response envelope parsing (response).

Source code in graflo/architecture/contract/bindings/connectors.py

class PaginationConfig(ConfigBaseModel):
    """Configuration for API pagination (contract-level, secret-free).

    Combines request construction (``request``) with response envelope parsing
    (``response``).
    """

    request: PaginationRequestConfig = Field(default_factory=PaginationRequestConfig)
    response: ApiResponseStructure = Field(default_factory=ApiResponseStructure)

Attributes¶

`request = Field(default_factory=PaginationRequestConfig)` `class-attribute` `instance-attribute` ¶

`response = Field(default_factory=ApiResponseStructure)` `class-attribute` `instance-attribute` ¶

`PaginationRequestConfig` ¶

Bases: ConfigBaseModel

Configuration for building paginated HTTP requests.

Source code in graflo/architecture/contract/bindings/connectors.py

class PaginationRequestConfig(ConfigBaseModel):
    """Configuration for building paginated HTTP requests."""

    strategy: Literal["offset", "page", "cursor"] = "offset"
    offset_param: str = "offset"
    limit_param: str | None = Field(
        default="limit",
        description=(
            "Query parameter name for page size (offset strategy only). "
            "The value sent is ``page_size``, not a total item cap. "
            "When ``None``, no page-size query parameter is sent."
        ),
    )
    cursor_param: str = "cursor"
    page_param: str = "page"
    per_page_param: str = Field(
        default="per_page",
        description=(
            "Query parameter name for page size (page strategy only). "
            "The value sent is ``page_size``, not a total item cap."
        ),
    )
    initial_offset: int = 0
    initial_page: int = 1
    initial_cursor: str | None = None
    page_size: int = Field(
        default=100,
        description=(
            "Records requested per HTTP page. Sent as the value of "
            "``limit_param`` (offset) or ``per_page_param`` (page) when those "
            "param names are set."
        ),
    )
    carry_params: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Query param name -> response dot path. After each page, values are "
            "read and sent on later requests (e.g. ``results_id: '0.results_id'``). "
            "When empty, known session-token fields are auto-detected from the "
            "first response."
        ),
    )

Attributes¶

carry_params = Field(default_factory=dict, description="Query param name -> response dot path. After each page, values are read and sent on later requests (e.g. ``results_id: '0.results_id'``). When empty, known session-token fields are auto-detected from the first response.") `class-attribute` `instance-attribute` ¶

`cursor_param = 'cursor'` `class-attribute` `instance-attribute` ¶

`initial_cursor = None` `class-attribute` `instance-attribute` ¶

`initial_offset = 0` `class-attribute` `instance-attribute` ¶

`initial_page = 1` `class-attribute` `instance-attribute` ¶

limit_param = Field(default='limit', description='Query parameter name for page size (offset strategy only). The value sent is ``page_size``, not a total item cap. When ``None``, no page-size query parameter is sent.') `class-attribute` `instance-attribute` ¶

`offset_param = 'offset'` `class-attribute` `instance-attribute` ¶

`page_param = 'page'` `class-attribute` `instance-attribute` ¶

page_size = Field(default=100, description='Records requested per HTTP page. Sent as the value of ``limit_param`` (offset) or ``per_page_param`` (page) when those param names are set.') `class-attribute` `instance-attribute` ¶

per_page_param = Field(default='per_page', description='Query parameter name for page size (page strategy only). The value sent is ``page_size``, not a total item cap.') `class-attribute` `instance-attribute` ¶

`strategy = 'offset'` `class-attribute` `instance-attribute` ¶

`ParentRequest` ¶

Bases: ConfigBaseModel

Make one request per record of another API connector.

Attributes:

Name	Type	Description
`connector`	`str`	Name of the parent API connector in the same bindings.
`fields`	`dict[str, str]`	Template variable -> dot path into a parent record. Each variable fills `{variable}` in the child's `path`, `params` and `body`; a parent record missing any of them is skipped.

Source code in graflo/architecture/contract/bindings/connectors.py

class ParentRequest(ConfigBaseModel):
    """Make one request per record of another API connector.

    Attributes:
        connector: Name of the parent API connector in the same bindings.
        fields: Template variable -> dot path into a parent record. Each
            variable fills ``{variable}`` in the child's ``path``, ``params``
            and ``body``; a parent record missing any of them is skipped.
    """

    connector: str = Field(..., min_length=1)
    fields: dict[str, str] = Field(..., min_length=1)

Attributes¶

`connector = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶

`fields = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶

`PushConnector` ¶

Bases: ResourceConnector

Connector for records pushed to a resource rather than fetched.

It has no address: the caller hands the records over at run time (RegistryBuilder.build(pushed=...)), keyed by the connector name. Declaring it keeps what a resource accepts versioned and reviewable like any other source.

Attributes:

Name	Type	Description
`kind`	`Literal['push']`	Always `push`; required, so the connector is never read as another kind.
`name`	`str`	Name the pushed records are keyed by.
`records_path`	`str \| None`	Dot path to the record list when a pushed item is an envelope rather than a record.
`row_annotations`	`dict[str, Any]`	Constant fields merged into every record (record wins).

Source code in graflo/architecture/contract/bindings/connectors.py

class PushConnector(ResourceConnector):
    """Connector for records pushed to a resource rather than fetched.

    It has no address: the caller hands the records over at run time
    (``RegistryBuilder.build(pushed=...)``), keyed by the connector name. Declaring
    it keeps what a resource accepts versioned and reviewable like any other
    source.

    Attributes:
        kind: Always ``push``; required, so the connector is never read as
            another kind.
        name: Name the pushed records are keyed by.
        records_path: Dot path to the record list when a pushed item is an
            envelope rather than a record.
        row_annotations: Constant fields merged into every record (record wins).
    """

    takes_default_conn_proxy: ClassVar[bool] = False

    kind: Literal["push"]
    # Narrowed from the base's optional name; validate_assignment refuses None.
    name: str = Field(..., min_length=1)  # ty: ignore[invalid-mutable-override]
    records_path: str | None = None

    def _hash_payload(self) -> dict[str, Any]:
        payload = super()._hash_payload()
        # It has no address: the name is what records are pushed to, so two
        # push connectors with different names are two sources.
        payload["name"] = self.name
        return payload

    def matches(self, resource_identifier: str) -> bool:
        """Match connector name or resource name."""
        return resource_identifier == self.name or (
            self.resource_name is not None and resource_identifier == self.resource_name
        )

    def bound_source_kind(self) -> BoundSourceKind:
        return BoundSourceKind.PUSH

Attributes¶

`kind` `instance-attribute` ¶

`name = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶

`records_path = None` `class-attribute` `instance-attribute` ¶

`takes_default_conn_proxy = False` `class-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    return BoundSourceKind.PUSH

`matches(resource_identifier)` ¶

Match connector name or resource name.

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Match connector name or resource name."""
    return resource_identifier == self.name or (
        self.resource_name is not None and resource_identifier == self.resource_name
    )

`ResourceConnector` ¶

Bases: ConfigBaseModel, ABC

Abstract base class for resource connectors (files or tables).

Provides common API for connector matching and resource identification. All concrete connector types inherit from this class.

Connectors only describe source-side matching/query behavior. Resource-to- connector linkage is handled by Bindings.

Source code in graflo/architecture/contract/bindings/connectors.py

class ResourceConnector(ConfigBaseModel, abc.ABC):
    """Abstract base class for resource connectors (files or tables).

    Provides common API for connector matching and resource identification.
    All concrete connector types inherit from this class.

    Connectors only describe source-side matching/query behavior. Resource-to-
    connector linkage is handled by ``Bindings``.
    """

    name: str | None = Field(
        default=None,
        description="Optional connector name used by top-level resource_connector mapping.",
    )
    resource_name: str | None = Field(
        default=None,
        description="Optional direct resource binding declared on the connector itself.",
    )
    hash: str = Field(
        default="",
        exclude=True,
        description="Deterministic internal connector id derived from defining fields.",
    )
    #: Whether the bindings' default ``conn_proxy`` applies to this connector.
    #: A connector that can be read without a connection only uses one wired
    #: to it explicitly.
    takes_default_conn_proxy: ClassVar[bool] = True
    row_annotations: dict[str, Any] = Field(
        default_factory=dict,
        description=(
            "Constant fields merged into every fetched row as defaults (response "
            "fields take priority). Implemented for :class:`APIConnector` and "
            ":class:`KafkaConnector`; other connector types reject non-empty values."
        ),
    )

    def _hash_payload(self) -> dict[str, Any]:
        payload = self.model_dump(
            mode="json",
            by_alias=True,
            exclude={"hash", "name", "resource_name"},
        )
        payload["_connector_type"] = type(self).__name__
        return payload

    @model_validator(mode="after")
    def _compute_hash(self) -> Self:
        object.__setattr__(self, "hash", suthing.stable_hash(self._hash_payload()))
        return self

    @abc.abstractmethod
    def matches(self, resource_identifier: str) -> bool:
        """Check if connector matches a resource identifier.

        Args:
            resource_identifier: Identifier to match (filename or table name)

        Returns:
            bool: True if connector matches
        """

    @abc.abstractmethod
    def bound_source_kind(self) -> BoundSourceKind:
        """Return the physical source kind for this connector."""

    def row_constants(self) -> list[dict[str, Any]]:
        """The constant fields this connector puts on its rows, one mapping per request.

        A row field of the same name wins over a constant, so these are what
        the connector fixes, not what every row ends up holding.
        """
        return [dict(self.row_annotations)]

Attributes¶

`hash = Field(default='', exclude=True, description='Deterministic internal connector id derived from defining fields.')` `class-attribute` `instance-attribute` ¶

`name = Field(default=None, description='Optional connector name used by top-level resource_connector mapping.')` `class-attribute` `instance-attribute` ¶

`resource_name = Field(default=None, description='Optional direct resource binding declared on the connector itself.')` `class-attribute` `instance-attribute` ¶

row_annotations = Field(default_factory=dict, description='Constant fields merged into every fetched row as defaults (response fields take priority). Implemented for :class:`APIConnector` and :class:`KafkaConnector`; other connector types reject non-empty values.') `class-attribute` `instance-attribute` ¶

`takes_default_conn_proxy = True` `class-attribute` ¶

Methods:¶

`bound_source_kind()` `abstractmethod` ¶

Return the physical source kind for this connector.

Source code in graflo/architecture/contract/bindings/connectors.py

@abc.abstractmethod
def bound_source_kind(self) -> BoundSourceKind:
    """Return the physical source kind for this connector."""

`matches(resource_identifier)` `abstractmethod` ¶

Check if connector matches a resource identifier.

Parameters:

Name	Type	Description	Default
`resource_identifier`	`str`	Identifier to match (filename or table name)	required

Returns:

Name	Type	Description
`bool`	`bool`	True if connector matches

Source code in graflo/architecture/contract/bindings/connectors.py

@abc.abstractmethod
def matches(self, resource_identifier: str) -> bool:
    """Check if connector matches a resource identifier.

    Args:
        resource_identifier: Identifier to match (filename or table name)

    Returns:
        bool: True if connector matches
    """

`row_constants()` ¶

The constant fields this connector puts on its rows, one mapping per request.

A row field of the same name wins over a constant, so these are what the connector fixes, not what every row ends up holding.

Source code in graflo/architecture/contract/bindings/connectors.py

def row_constants(self) -> list[dict[str, Any]]:
    """The constant fields this connector puts on its rows, one mapping per request.

    A row field of the same name wins over a constant, so these are what
    the connector fixes, not what every row ends up holding.
    """
    return [dict(self.row_annotations)]

`SparqlConnector` ¶

Bases: TemplatedConnector

Connector for matching SPARQL / RDF data sources.

Each request targets instances of a single rdf:Class; with values the connector makes one request per value (see :class:TemplatedConnector). rdf_class and graph_uri are address templates, sparql_query a query text. It can be backed either by a remote SPARQL endpoint (Fuseki, Blazegraph, ...) or by a local RDF file parsed with rdflib.

Attributes:

Name	Type	Description
`rdf_class`	`str`	Full URI of the `rdf:Class` whose instances this connector fetches (e.g. `"http://example.org/Person"`), or a template of one (`"http://example.org/{cls}"`).
`endpoint_url`	`str \| None`	SPARQL query endpoint URL. When set, instances are fetched via HTTP. When `None` the connector is for local file mode.
`graph_uri`	`str \| None`	Named-graph URI to restrict the query to (optional).
`sparql_query`	`str \| None`	Custom SPARQL `SELECT` query override. When provided the auto-generated per-class query is skipped.
`rdf_file`	`Path \| None`	Path to a local RDF file (`.ttl`, `.rdf`, `.n3`, `.jsonld`). Mutually exclusive with endpoint_url.
`same_as`	`Literal['collapse', 'keep']`	`collapse` (default) reads IRIs joined by `owl:sameAs` as one record under the smallest IRI, the others listed in `_same_as`. `keep` reads the statements as a property.

Source code in graflo/architecture/contract/bindings/connectors.py

class SparqlConnector(TemplatedConnector):
    """Connector for matching SPARQL / RDF data sources.

    Each request targets instances of a single ``rdf:Class``; with ``values``
    the connector makes one request per value (see :class:`TemplatedConnector`).
    ``rdf_class`` and ``graph_uri`` are address templates, ``sparql_query`` a
    query text. It can be backed either by a remote SPARQL endpoint (Fuseki,
    Blazegraph, ...) or by a local RDF file parsed with *rdflib*.

    Attributes:
        rdf_class: Full URI of the ``rdf:Class`` whose instances this connector
            fetches (e.g. ``"http://example.org/Person"``), or a template of
            one (``"http://example.org/{cls}"``).
        endpoint_url: SPARQL query endpoint URL.  When set, instances are
            fetched via HTTP.  When ``None`` the connector is for local file mode.
        graph_uri: Named-graph URI to restrict the query to (optional).
        sparql_query: Custom SPARQL ``SELECT`` query override.  When provided
            the auto-generated per-class query is skipped.
        rdf_file: Path to a local RDF file (``.ttl``, ``.rdf``, ``.n3``,
            ``.jsonld``).  Mutually exclusive with *endpoint_url*.
        same_as: ``collapse`` (default) reads IRIs joined by ``owl:sameAs`` as
            one record under the smallest IRI, the others listed in
            ``_same_as``. ``keep`` reads the statements as a property.
    """

    rdf_class: str = Field(
        ..., description="URI of the rdf:Class to fetch instances of"
    )
    endpoint_url: str | None = Field(
        default=None, description="SPARQL query endpoint URL"
    )
    graph_uri: str | None = Field(
        default=None, description="Named graph URI (optional)"
    )
    sparql_query: str | None = Field(
        default=None, description="Custom SPARQL query override"
    )
    rdf_file: pathlib.Path | None = Field(
        default=None, description="Path to a local RDF file"
    )
    same_as: Literal["collapse", "keep"] = Field(
        default="collapse",
        description="How owl:sameAs statements between IRIs are read.",
    )
    typed_objects: list[str] = Field(
        default_factory=list,
        description=(
            "Properties (local names) whose objects are also read by class, as "
            "``<property>@<Class>``: for a property with several ranges."
        ),
    )

    def _hash_payload(self) -> dict[str, Any]:
        payload = super()._hash_payload()
        # Left out at their defaults, so a connector that does not set them
        # keeps the hash it is referred to by.
        if self.same_as == "collapse":
            del payload["same_as"]
        if not self.typed_objects:
            del payload["typed_objects"]
        return payload

    @model_validator(mode="after")
    def _reject_row_annotations(self) -> Self:
        # TODO: implement row_annotations for SparqlConnector row payloads.
        if self.row_annotations:
            raise ValueError("row_annotations is not implemented for SparqlConnector")
        return self

    def _address_templates(self) -> list[Any]:
        return [self.rdf_class, self.graph_uri]

    def _text_templates(self) -> list[Any]:
        return [self.sparql_query]

    def _label(self) -> str:
        return self.name or self.rdf_class

    def rendered(self, values: Mapping[str, Any]) -> Self:
        return self.model_copy(
            update={
                "rdf_class": render_text(self.rdf_class, values),
                "graph_uri": None
                if self.graph_uri is None
                else render_text(self.graph_uri, values),
                "sparql_query": None
                if self.sparql_query is None
                else render_text(self.sparql_query, values),
                "values": None,
            }
        )

    def matches(self, resource_identifier: str) -> bool:
        """Match by the local name (fragment) of the rdf:Class URI.

        Args:
            resource_identifier: Identifier to match against

        Returns:
            True when *resource_identifier* equals the class local name
        """
        local_name = self.rdf_class.rsplit("#", 1)[-1].rsplit("/", 1)[-1]
        return resource_identifier == local_name

    def bound_source_kind(self) -> BoundSourceKind:
        """Return ``BoundSourceKind.SPARQL``."""
        return BoundSourceKind.SPARQL

    def build_select_query(self) -> str:
        """Build a SPARQL SELECT query for instances of ``rdf_class``.

        If *sparql_query* is set it is returned as-is.  Otherwise a simple
        per-class query is generated::

            SELECT ?s ?p ?o WHERE {
              ?s a <rdf_class> .
              ?s ?p ?o .
            }

        Returns:
            SPARQL query string
        """
        if self.sparql_query:
            return self.sparql_query

        graph_open = f"GRAPH <{self.graph_uri}> {{" if self.graph_uri else ""
        graph_close = "}" if self.graph_uri else ""

        return (
            "SELECT ?s ?p ?o WHERE { "
            f"{graph_open} "
            f"?s a <{self.rdf_class}> . "
            f"?s ?p ?o . "
            f"{graph_close} "
            "}"
        )

Attributes¶

`endpoint_url = Field(default=None, description='SPARQL query endpoint URL')` `class-attribute` `instance-attribute` ¶

`graph_uri = Field(default=None, description='Named graph URI (optional)')` `class-attribute` `instance-attribute` ¶

`rdf_class = Field(..., description='URI of the rdf:Class to fetch instances of')` `class-attribute` `instance-attribute` ¶

`rdf_file = Field(default=None, description='Path to a local RDF file')` `class-attribute` `instance-attribute` ¶

`same_as = Field(default='collapse', description='How owl:sameAs statements between IRIs are read.')` `class-attribute` `instance-attribute` ¶

`sparql_query = Field(default=None, description='Custom SPARQL query override')` `class-attribute` `instance-attribute` ¶

typed_objects = Field(default_factory=list, description='Properties (local names) whose objects are also read by class, as ``<property>@<Class>``: for a property with several ranges.') `class-attribute` `instance-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

Return BoundSourceKind.SPARQL.

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    """Return ``BoundSourceKind.SPARQL``."""
    return BoundSourceKind.SPARQL

`build_select_query()` ¶

Build a SPARQL SELECT query for instances of rdf_class.

If sparql_query is set it is returned as-is. Otherwise a simple per-class query is generated::

SELECT ?s ?p ?o WHERE {
  ?s a <rdf_class> .
  ?s ?p ?o .
}

Returns:

Type	Description
`str`	SPARQL query string

Source code in graflo/architecture/contract/bindings/connectors.py

def build_select_query(self) -> str:
    """Build a SPARQL SELECT query for instances of ``rdf_class``.

    If *sparql_query* is set it is returned as-is.  Otherwise a simple
    per-class query is generated::

        SELECT ?s ?p ?o WHERE {
          ?s a <rdf_class> .
          ?s ?p ?o .
        }

    Returns:
        SPARQL query string
    """
    if self.sparql_query:
        return self.sparql_query

    graph_open = f"GRAPH <{self.graph_uri}> {{" if self.graph_uri else ""
    graph_close = "}" if self.graph_uri else ""

    return (
        "SELECT ?s ?p ?o WHERE { "
        f"{graph_open} "
        f"?s a <{self.rdf_class}> . "
        f"?s ?p ?o . "
        f"{graph_close} "
        "}"
    )

`matches(resource_identifier)` ¶

Match by the local name (fragment) of the rdf:Class URI.

Parameters:

Name	Type	Description	Default
`resource_identifier`	`str`	Identifier to match against	required

Returns:

Type	Description
`bool`	True when resource_identifier equals the class local name

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Match by the local name (fragment) of the rdf:Class URI.

    Args:
        resource_identifier: Identifier to match against

    Returns:
        True when *resource_identifier* equals the class local name
    """
    local_name = self.rdf_class.rsplit("#", 1)[-1].rsplit("/", 1)[-1]
    return resource_identifier == local_name

`rendered(values)` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def rendered(self, values: Mapping[str, Any]) -> Self:
    return self.model_copy(
        update={
            "rdf_class": render_text(self.rdf_class, values),
            "graph_uri": None
            if self.graph_uri is None
            else render_text(self.graph_uri, values),
            "sparql_query": None
            if self.sparql_query is None
            else render_text(self.sparql_query, values),
            "values": None,
        }
    )

`TableConnector` ¶

Bases: ResourceConnector

Connector for matching database tables.

Supports simple single-table queries as well as multi-table JOINs and pushdown filters via FilterExpression.

Attributes:

Name	Type	Description
`table_name`	`str`	Exact table name or regex pattern
`schema_name`	`str \| None`	Schema name (optional, defaults to public)
`database`	`str \| None`	Database name (optional)
`time_filter`	`ColumnTimeFilter \| None`	Optional structured filter on a date/time column, rendered via :class:`~graflo.filter.onto.FilterExpression` in SQL.
`filters`	`list[Any]`	General-purpose pushdown filters rendered as SQL WHERE fragments.
`joins`	`list[JoinClause]`	Multi-table JOIN specifications (auto-generated or explicit).
`base_alias`	`str`	SQL alias for the base table when `joins` is non-empty.
`select_columns`	`list[str] \| None`	Explicit SELECT column list. None means `*` for the base table (plus aliased columns from joins).

Source code in graflo/architecture/contract/bindings/connectors.py

class TableConnector(ResourceConnector):
    """Connector for matching database tables.

    Supports simple single-table queries as well as multi-table JOINs and
    pushdown filters via ``FilterExpression``.

    Attributes:
        table_name: Exact table name or regex pattern
        schema_name: Schema name (optional, defaults to public)
        database: Database name (optional)
        time_filter: Optional structured filter on a date/time column, rendered
            via :class:`~graflo.filter.onto.FilterExpression` in SQL.
        filters: General-purpose pushdown filters rendered as SQL WHERE fragments.
        joins: Multi-table JOIN specifications (auto-generated or explicit).
        base_alias: SQL alias for the base table when ``joins`` is non-empty.
        select_columns: Explicit SELECT column list. None means ``*`` for the
            base table (plus aliased columns from joins).
    """

    table_name: str = Field(
        default="", validation_alias=AliasChoices("table_name", "table")
    )
    schema_name: str | None = Field(
        default=None, validation_alias=AliasChoices("schema_name", "schema")
    )
    database: str | None = None
    time_filter: ColumnTimeFilter | None = None
    filters: list[Any] = Field(
        default_factory=list,
        description="Pushdown FilterExpression filters (rendered to SQL WHERE).",
    )
    joins: list[JoinClause] = Field(
        default_factory=list,
        description="JOIN clauses for multi-table queries.",
    )
    base_alias: str = Field(
        default="base",
        description="SQL alias for the base table row when joins are present.",
    )
    select_columns: list[str] | None = Field(
        default=None,
        description="Explicit SELECT columns. None = SELECT * (plus join aliases).",
    )
    view: Any = Field(
        default=None,
        description="SelectSpec or dict for declarative view (alternative to table+joins+filters).",
    )

    @field_validator("filters", mode="before")
    @classmethod
    def _coerce_filters(cls, v: Any) -> list[Any]:
        from graflo.filter.onto import parse_filter_expression

        if v is None:
            return []
        if not isinstance(v, list):
            raise ValueError("filters must be a list")
        result: list[Any] = []
        for i, item in enumerate(v):
            try:
                result.append(parse_filter_expression(item))
            except (ValueError, ValidationError) as e:
                raise ValueError(f"filters[{i}]: {e}") from e
        return result

    @field_validator("view", mode="before")
    @classmethod
    def _coerce_view(cls, v: Any) -> Any:
        if v is None:
            return None
        if isinstance(v, dict):
            from graflo.filter.select import SelectSpec

            return SelectSpec.from_dict(v)
        return v

    @model_validator(mode="after")
    def _validate_table_connector(self) -> Self:
        """Validate table_name and join wiring."""
        if not self.table_name:
            raise ValueError("table_name is required for TableConnector")
        if not _BASE_TABLE_ALIAS_IDENT.match(self.base_alias):
            raise ValueError(
                "base_alias must be a simple SQL identifier "
                "(ASCII letter, digit, underscore)"
            )
        join_aliases = {jc.alias or jc.table for jc in self.joins}
        if self.base_alias in join_aliases:
            raise ValueError(
                f"base_alias {self.base_alias!r} conflicts with a join alias "
                f"{sorted(join_aliases)}"
            )
        if self.row_annotations:
            raise ValueError("row_annotations is not implemented for TableConnector")
        return self

    @property
    def date_field(self) -> str | None:
        """Column used for time filtering, if any (compat alias for ``time_filter.column``)."""
        return self.time_filter.column if self.time_filter else None

    def matches(self, resource_identifier: str) -> bool:
        """Check if connector matches a table name.

        Args:
            resource_identifier: Table name to match (format: schema.table or just table)

        Returns:
            bool: True if connector matches
        """
        if not self.table_name:
            return False

        # Compile regex expression
        if self.table_name.startswith("^") or self.table_name.endswith("$"):
            # Already a regex expression
            compiled_regex = re.compile(self.table_name)
        else:
            # Exact match expression
            compiled_regex = re.compile(f"^{re.escape(self.table_name)}$")

        # Check if resource_identifier matches
        if compiled_regex.match(resource_identifier):
            return True

        # If schema_name is specified, also check schema.table format
        if self.schema_name:
            full_name = f"{self.schema_name}.{resource_identifier}"
            if compiled_regex.match(full_name):
                return True

        return False

    def bound_source_kind(self) -> BoundSourceKind:
        return BoundSourceKind.SQL_TABLE

    def where_conditions(
        self,
        base_alias: str | None = None,
        extra_filters: Sequence[FilterExpression] = (),
    ) -> list[str]:
        """The connector's SQL conditions, each safe as one operand of ``AND``.

        In order: the time filter, the declared ``filters``, then
        *extra_filters* (conditions supplied for one run). With *base_alias*,
        unqualified columns are qualified with it.
        """
        from graflo.filter.onto import render_conjunct
        from graflo.onto import ExpressionFlavor

        raw: list[Any] = []
        if self.time_filter is not None:
            raw.append(self.time_filter.as_filter_expression())
        raw.extend(self.filters)
        raw.extend(extra_filters)

        conditions: list[str] = []
        for item in raw:
            filt_expr = self._coerce_filter_expression(item, base_alias)
            if filt_expr is None:
                continue
            rendered = render_conjunct(filt_expr, kind=ExpressionFlavor.SQL)
            if rendered:
                conditions.append(rendered)
        return conditions

    def build_where_clause(
        self,
        base_alias: str | None = None,
        extra_filters: Sequence[FilterExpression] = (),
    ) -> str:
        """Build SQL WHERE clause from time filter **and** general filters.

        Returns:
            WHERE clause string (without the WHERE keyword) or empty string if no filters
        """
        return " AND ".join(self.where_conditions(base_alias, extra_filters))

    def build_query(
        self,
        effective_schema: str | None = None,
        *,
        extra_filters: Sequence[FilterExpression] = (),
    ) -> str:
        """Build a complete SQL SELECT query.

        When ``view`` is set, delegates to ``view.build_sql()``. Otherwise
        incorporates the base table, any JoinClauses, explicit select_columns,
        time_filter, and FilterExpression filters.

        Args:
            effective_schema: Schema to use if ``self.schema_name`` is None.
            extra_filters: Conditions for this run only, such as a date range
                supplied at ingestion time. They are ANDed with the declared
                ones and qualified like them.

        Returns:
            Complete SQL query string.
        """
        schema = self.schema_name or effective_schema or "public"
        if self.view is not None:
            from graflo.filter.select import SelectSpec

            if isinstance(self.view, SelectSpec):
                return self.view.build_sql(
                    schema=schema,
                    base_table=self.table_name,
                    extra_conditions=self.where_conditions(
                        base_alias=self.view.effective_base_alias(),
                        extra_filters=extra_filters,
                    ),
                )
        base_alias = self.base_alias if self.joins else None
        base_ref = f'"{schema}"."{self.table_name}"'
        if base_alias:
            base_ref_aliased = f"{base_ref} {base_alias}"
        else:
            base_ref_aliased = base_ref

        # --- SELECT ---
        select_parts: list[str] = []
        if self.select_columns is not None:
            select_parts = list(self.select_columns)
        elif self.joins:
            select_parts.append(f"{base_alias}.*")
            for jc in self.joins:
                alias = jc.alias or jc.table
                jc_schema = jc.schema_name or schema
                if jc.select_fields is not None:
                    for col in jc.select_fields:
                        select_parts.append(f'{alias}."{col}" AS "{alias}__{col}"')
                else:
                    select_parts.append(f"{alias}.*")
        else:
            select_parts.append("*")

        select_clause = ", ".join(select_parts)

        # --- FROM + JOINs ---
        from_clause = base_ref_aliased
        for jc in self.joins:
            jc_schema = jc.schema_name or schema
            alias = jc.alias or jc.table
            join_ref = f'"{jc_schema}"."{jc.table}"'
            left_col = (
                f'{base_alias}."{jc.on_self}"' if base_alias else f'"{jc.on_self}"'
            )
            right_col = f'{alias}."{jc.on_other}"'
            from_clause += (
                f" {jc.join_type} JOIN {join_ref} {alias} ON {left_col} = {right_col}"
            )

        query = f"SELECT {select_clause} FROM {from_clause}"

        # --- WHERE ---
        where = self.build_where_clause(
            base_alias=base_alias, extra_filters=extra_filters
        )
        if where:
            query += f" WHERE {where}"

        return query

    def source_column(self, name: str) -> str | None:
        """The column the query's output *name* reads, as a filter field.

        A base column is returned bare (conditions qualify it with the base
        alias), a joined one as ``alias.column``. ``None`` when the query's
        output names are not known: raw ``select_columns``, or a join that
        selects all its columns.
        """
        if self.view is not None:
            from graflo.filter.select import SelectSpec

            return (
                self.view.source_column(name)
                if isinstance(self.view, SelectSpec)
                else None
            )
        if self.select_columns is not None:
            return None
        for jc in self.joins:
            if jc.select_fields is None:
                return None
            alias = jc.alias or jc.table
            for col in jc.select_fields:
                if name == f"{alias}__{col}":
                    return f"{alias}.{col}"
        return name

    @staticmethod
    def _qualified_column_ref(column: str, base_alias: str | None) -> str:
        if base_alias:
            return f'{base_alias}."{column}"'
        return f'"{column}"'

    @classmethod
    def _qualify_filter_payload(
        cls, payload: dict[str, Any], base_alias: str | None
    ) -> dict[str, Any]:
        qualified = dict(payload)
        if base_alias is None:
            return qualified
        if qualified.get("kind") == "leaf":
            field = qualified.get("field")
            if isinstance(field, str) and "." not in field:
                qualified["field"] = f"{base_alias}.{field}"
            return qualified
        deps = qualified.get("deps")
        if isinstance(deps, list):
            qualified["deps"] = [
                cls._qualify_filter_payload(dep, base_alias)
                if isinstance(dep, dict)
                else dep
                for dep in deps
            ]
        return qualified

    @classmethod
    def _coerce_filter_expression(
        cls, raw_filter: Any, base_alias: str | None
    ) -> FilterExpression | None:
        from graflo.filter.onto import parse_filter_expression

        if raw_filter is None:
            return None
        expr = parse_filter_expression(raw_filter)
        if base_alias is None:
            return expr
        payload = expr.model_dump(mode="python")
        return parse_filter_expression(cls._qualify_filter_payload(payload, base_alias))

Attributes¶

`base_alias = Field(default='base', description='SQL alias for the base table row when joins are present.')` `class-attribute` `instance-attribute` ¶

`database = None` `class-attribute` `instance-attribute` ¶

`date_field` `property` ¶

Column used for time filtering, if any (compat alias for time_filter.column).

`filters = Field(default_factory=list, description='Pushdown FilterExpression filters (rendered to SQL WHERE).')` `class-attribute` `instance-attribute` ¶

`joins = Field(default_factory=list, description='JOIN clauses for multi-table queries.')` `class-attribute` `instance-attribute` ¶

`schema_name = Field(default=None, validation_alias=AliasChoices('schema_name', 'schema'))` `class-attribute` `instance-attribute` ¶

`select_columns = Field(default=None, description='Explicit SELECT columns. None = SELECT * (plus join aliases).')` `class-attribute` `instance-attribute` ¶

`table_name = Field(default='', validation_alias=AliasChoices('table_name', 'table'))` `class-attribute` `instance-attribute` ¶

`time_filter = None` `class-attribute` `instance-attribute` ¶

`view = Field(default=None, description='SelectSpec or dict for declarative view (alternative to table+joins+filters).')` `class-attribute` `instance-attribute` ¶

Methods:¶

`bound_source_kind()` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def bound_source_kind(self) -> BoundSourceKind:
    return BoundSourceKind.SQL_TABLE

`build_query(effective_schema=None, *, extra_filters=())` ¶

Build a complete SQL SELECT query.

When view is set, delegates to view.build_sql(). Otherwise incorporates the base table, any JoinClauses, explicit select_columns, time_filter, and FilterExpression filters.

Parameters:

Name	Type	Description	Default
`effective_schema`	`str \| None`	Schema to use if `self.schema_name` is None.	`None`
`extra_filters`	`Sequence[FilterExpression]`	Conditions for this run only, such as a date range supplied at ingestion time. They are ANDed with the declared ones and qualified like them.	`()`

Returns:

Type	Description
`str`	Complete SQL query string.

Source code in graflo/architecture/contract/bindings/connectors.py

def build_query(
    self,
    effective_schema: str | None = None,
    *,
    extra_filters: Sequence[FilterExpression] = (),
) -> str:
    """Build a complete SQL SELECT query.

    When ``view`` is set, delegates to ``view.build_sql()``. Otherwise
    incorporates the base table, any JoinClauses, explicit select_columns,
    time_filter, and FilterExpression filters.

    Args:
        effective_schema: Schema to use if ``self.schema_name`` is None.
        extra_filters: Conditions for this run only, such as a date range
            supplied at ingestion time. They are ANDed with the declared
            ones and qualified like them.

    Returns:
        Complete SQL query string.
    """
    schema = self.schema_name or effective_schema or "public"
    if self.view is not None:
        from graflo.filter.select import SelectSpec

        if isinstance(self.view, SelectSpec):
            return self.view.build_sql(
                schema=schema,
                base_table=self.table_name,
                extra_conditions=self.where_conditions(
                    base_alias=self.view.effective_base_alias(),
                    extra_filters=extra_filters,
                ),
            )
    base_alias = self.base_alias if self.joins else None
    base_ref = f'"{schema}"."{self.table_name}"'
    if base_alias:
        base_ref_aliased = f"{base_ref} {base_alias}"
    else:
        base_ref_aliased = base_ref

    # --- SELECT ---
    select_parts: list[str] = []
    if self.select_columns is not None:
        select_parts = list(self.select_columns)
    elif self.joins:
        select_parts.append(f"{base_alias}.*")
        for jc in self.joins:
            alias = jc.alias or jc.table
            jc_schema = jc.schema_name or schema
            if jc.select_fields is not None:
                for col in jc.select_fields:
                    select_parts.append(f'{alias}."{col}" AS "{alias}__{col}"')
            else:
                select_parts.append(f"{alias}.*")
    else:
        select_parts.append("*")

    select_clause = ", ".join(select_parts)

    # --- FROM + JOINs ---
    from_clause = base_ref_aliased
    for jc in self.joins:
        jc_schema = jc.schema_name or schema
        alias = jc.alias or jc.table
        join_ref = f'"{jc_schema}"."{jc.table}"'
        left_col = (
            f'{base_alias}."{jc.on_self}"' if base_alias else f'"{jc.on_self}"'
        )
        right_col = f'{alias}."{jc.on_other}"'
        from_clause += (
            f" {jc.join_type} JOIN {join_ref} {alias} ON {left_col} = {right_col}"
        )

    query = f"SELECT {select_clause} FROM {from_clause}"

    # --- WHERE ---
    where = self.build_where_clause(
        base_alias=base_alias, extra_filters=extra_filters
    )
    if where:
        query += f" WHERE {where}"

    return query

`build_where_clause(base_alias=None, extra_filters=())` ¶

Build SQL WHERE clause from time filter and general filters.

Returns:

Type	Description
`str`	WHERE clause string (without the WHERE keyword) or empty string if no filters

Source code in graflo/architecture/contract/bindings/connectors.py

def build_where_clause(
    self,
    base_alias: str | None = None,
    extra_filters: Sequence[FilterExpression] = (),
) -> str:
    """Build SQL WHERE clause from time filter **and** general filters.

    Returns:
        WHERE clause string (without the WHERE keyword) or empty string if no filters
    """
    return " AND ".join(self.where_conditions(base_alias, extra_filters))

`matches(resource_identifier)` ¶

Check if connector matches a table name.

Parameters:

Name	Type	Description	Default
`resource_identifier`	`str`	Table name to match (format: schema.table or just table)	required

Returns:

Name	Type	Description
`bool`	`bool`	True if connector matches

Source code in graflo/architecture/contract/bindings/connectors.py

def matches(self, resource_identifier: str) -> bool:
    """Check if connector matches a table name.

    Args:
        resource_identifier: Table name to match (format: schema.table or just table)

    Returns:
        bool: True if connector matches
    """
    if not self.table_name:
        return False

    # Compile regex expression
    if self.table_name.startswith("^") or self.table_name.endswith("$"):
        # Already a regex expression
        compiled_regex = re.compile(self.table_name)
    else:
        # Exact match expression
        compiled_regex = re.compile(f"^{re.escape(self.table_name)}$")

    # Check if resource_identifier matches
    if compiled_regex.match(resource_identifier):
        return True

    # If schema_name is specified, also check schema.table format
    if self.schema_name:
        full_name = f"{self.schema_name}.{resource_identifier}"
        if compiled_regex.match(full_name):
            return True

    return False

`source_column(name)` ¶

The column the query's output name reads, as a filter field.

A base column is returned bare (conditions qualify it with the base alias), a joined one as alias.column. None when the query's output names are not known: raw select_columns, or a join that selects all its columns.

Source code in graflo/architecture/contract/bindings/connectors.py

def source_column(self, name: str) -> str | None:
    """The column the query's output *name* reads, as a filter field.

    A base column is returned bare (conditions qualify it with the base
    alias), a joined one as ``alias.column``. ``None`` when the query's
    output names are not known: raw ``select_columns``, or a join that
    selects all its columns.
    """
    if self.view is not None:
        from graflo.filter.select import SelectSpec

        return (
            self.view.source_column(name)
            if isinstance(self.view, SelectSpec)
            else None
        )
    if self.select_columns is not None:
        return None
    for jc in self.joins:
        if jc.select_fields is None:
            return None
        alias = jc.alias or jc.table
        for col in jc.select_fields:
            if name == f"{alias}__{col}":
                return f"{alias}.{col}"
    return name

`where_conditions(base_alias=None, extra_filters=())` ¶

The connector's SQL conditions, each safe as one operand of AND.

In order: the time filter, the declared filters, then extra_filters (conditions supplied for one run). With base_alias, unqualified columns are qualified with it.

Source code in graflo/architecture/contract/bindings/connectors.py

def where_conditions(
    self,
    base_alias: str | None = None,
    extra_filters: Sequence[FilterExpression] = (),
) -> list[str]:
    """The connector's SQL conditions, each safe as one operand of ``AND``.

    In order: the time filter, the declared ``filters``, then
    *extra_filters* (conditions supplied for one run). With *base_alias*,
    unqualified columns are qualified with it.
    """
    from graflo.filter.onto import render_conjunct
    from graflo.onto import ExpressionFlavor

    raw: list[Any] = []
    if self.time_filter is not None:
        raw.append(self.time_filter.as_filter_expression())
    raw.extend(self.filters)
    raw.extend(extra_filters)

    conditions: list[str] = []
    for item in raw:
        filt_expr = self._coerce_filter_expression(item, base_alias)
        if filt_expr is None:
            continue
        rendered = render_conjunct(filt_expr, kind=ExpressionFlavor.SQL)
        if rendered:
            conditions.append(rendered)
    return conditions

`TemplatedConnector` ¶

Bases: ResourceConnector, ABC

A connector whose request fields may hold {name} variables.

values enumerates variables. The connector then makes one request per combination of their values -- in declaration order, the last variable varying fastest -- with the variables filled in, and every row carries the variables as fields unless it has a field of the same name. Which record becomes what is the resource's to say (a vertex_router on that field); the connector names no vertex.

Template fields are of two kinds. In an address (a URL path, a query parameter, an IRI) every {name} must be a declared variable. In a query text (a request body, a SPARQL query) braces are syntax, so only declared variables are filled and every other brace pair is literal.

Attributes:

Name	Type	Description
`values`	`dict[str, list[TemplateValue]] \| None`	Variable name -> the values it takes, in request order. Every variable must be used in a template field.

Source code in graflo/architecture/contract/bindings/connectors.py

class TemplatedConnector(ResourceConnector, abc.ABC):
    """A connector whose request fields may hold ``{name}`` variables.

    ``values`` enumerates variables. The connector then makes one request per
    combination of their values -- in declaration order, the last variable
    varying fastest -- with the variables filled in, and every row carries the
    variables as fields unless it has a field of the same name. Which record
    becomes what is the resource's to say (a ``vertex_router`` on that field);
    the connector names no vertex.

    Template fields are of two kinds. In an *address* (a URL path, a query
    parameter, an IRI) every ``{name}`` must be a declared variable. In a
    *query text* (a request body, a SPARQL query) braces are syntax, so only
    declared variables are filled and every other brace pair is literal.

    Attributes:
        values: Variable name -> the values it takes, in request order. Every
            variable must be used in a template field.
    """

    values: dict[str, list[TemplateValue]] | None = Field(
        default=None,
        description=(
            "Variable -> values; one request per combination, each row carrying "
            "the variables as fields."
        ),
    )

    @abc.abstractmethod
    def _address_templates(self) -> list[Any]:
        """Template fields in which every ``{name}`` must be declared."""

    @abc.abstractmethod
    def _text_templates(self) -> list[Any]:
        """Template fields in which only declared variables are filled."""

    def _bound_variables(self) -> set[str]:
        """Variables filled other than from ``values``."""
        return set()

    @abc.abstractmethod
    def _label(self) -> str:
        """How an error names this connector."""

    @abc.abstractmethod
    def rendered(self, values: Mapping[str, Any]) -> Self:
        """A copy for one request: template fields filled from *values*, no ``values``.

        The copy keeps this connector's name and hash, so its connection and
        bindings still resolve to it.
        """

    @field_validator("values")
    @classmethod
    def _check_values(
        cls, v: dict[str, list[TemplateValue]] | None
    ) -> dict[str, list[TemplateValue]] | None:
        if v is None:
            return v
        if not v:
            raise ValueError("values declares no variable; leave it unset")
        for name, items in v.items():
            if not _VARIABLE_NAME.match(name):
                raise ValueError(
                    f"values: {name!r} is not a variable name (a letter or "
                    "underscore, then letters, digits or underscores)"
                )
            if not items:
                raise ValueError(f"values: {name!r} lists no value")
            seen: list[tuple[type, TemplateValue]] = []
            for item in items:
                if (type(item), item) in seen:
                    raise ValueError(f"values: {name!r} lists {item!r} twice")
                seen.append((type(item), item))
        return v

    @model_validator(mode="after")
    def _check_variables(self) -> Self:
        enumerated = set(self.values or {})
        bound = self._bound_variables()
        if both := enumerated & bound:
            raise ValueError(
                f"connector '{self._label()}' declares {sorted(both)} both in "
                "values and as a parent field"
            )
        declared = enumerated | bound
        undeclared = template_variables(*self._address_templates()) - declared
        if undeclared:
            raise ValueError(
                f"connector '{self._label()}' uses template variables "
                f"{sorted(undeclared)} that neither values nor parent.fields declares"
            )
        used = template_variables(*self._address_templates(), *self._text_templates())
        if unused := enumerated - used:
            raise ValueError(
                f"connector '{self._label()}' declares values for {sorted(unused)} "
                "that no template field uses"
            )
        if clash := enumerated & set(self.row_annotations):
            raise ValueError(
                f"connector '{self._label()}' sets {sorted(clash)} both in values "
                "and in row_annotations"
            )
        return self

    def value_combinations(self) -> list[dict[str, Any]]:
        """One mapping per request, in request order; ``[{}]`` without ``values``."""
        if not self.values:
            return [{}]
        names = list(self.values)
        return [
            dict(zip(names, combination, strict=True))
            for combination in itertools.product(*(self.values[n] for n in names))
        ]

    def row_constants(self) -> list[dict[str, Any]]:
        return [
            {**self.row_annotations, **combination}
            for combination in self.value_combinations()
        ]

    def _hash_payload(self) -> dict[str, Any]:
        payload = super()._hash_payload()
        if self.values is None:
            # Unset, it must not move the hash of connectors declared before it.
            payload.pop("values", None)
        return payload

Attributes¶

`values = Field(default=None, description='Variable -> values; one request per combination, each row carrying the variables as fields.')` `class-attribute` `instance-attribute` ¶

Methods:¶

`rendered(values)` `abstractmethod` ¶

A copy for one request: template fields filled from values, no values.

The copy keeps this connector's name and hash, so its connection and bindings still resolve to it.

Source code in graflo/architecture/contract/bindings/connectors.py

@abc.abstractmethod
def rendered(self, values: Mapping[str, Any]) -> Self:
    """A copy for one request: template fields filled from *values*, no ``values``.

    The copy keeps this connector's name and hash, so its connection and
    bindings still resolve to it.
    """

`row_constants()` ¶

Source code in graflo/architecture/contract/bindings/connectors.py

def row_constants(self) -> list[dict[str, Any]]:
    return [
        {**self.row_annotations, **combination}
        for combination in self.value_combinations()
    ]

`value_combinations()` ¶

One mapping per request, in request order; [{}] without values.

Source code in graflo/architecture/contract/bindings/connectors.py

def value_combinations(self) -> list[dict[str, Any]]:
    """One mapping per request, in request order; ``[{}]`` without ``values``."""
    if not self.values:
        return [{}]
    names = list(self.values)
    return [
        dict(zip(names, combination, strict=True))
        for combination in itertools.product(*(self.values[n] for n in names))
    ]

`WatermarkConfig` ¶

Bases: ConfigBaseModel

Where an API source left off, and how to ask it to resume from there.

The source tracks the largest value of field over the rows it yields. Values compare as numbers when both parse as numbers, and as text otherwise, so timestamps must be written in one sortable form (ISO 8601 in one zone). The caller stores that value and renders it into the next run's query parameters; GraFlo keeps no run state itself.

Attributes:

Name	Type	Description
`field`	`str`	Dot path of the value on each row.
`param`	`str`	Query parameter that carries the value on the next run.
`template`	`str \| None`	Parameter text with a `{value}` placeholder; `None` sends the bare value.
`format`	`str \| None`	strftime pattern the value is rendered with, in UTC, before the template, for an API whose query language takes another time form than its records carry (`"%Y-%m-%d %H:%M"`). The stored value is read as ISO 8601 (`Z`, `+00:00` or `+0000` offsets, fractional seconds; no offset means UTC), or as epoch seconds when it is a number or a string of one. `None` renders it unchanged.

Source code in graflo/architecture/contract/bindings/connectors.py

class WatermarkConfig(ConfigBaseModel):
    """Where an API source left off, and how to ask it to resume from there.

    The source tracks the largest value of ``field`` over the rows it yields.
    Values compare as numbers when both parse as numbers, and as text
    otherwise, so timestamps must be written in one sortable form (ISO 8601 in
    one zone). The caller stores that value and renders it into the next run's
    query parameters; GraFlo keeps no run state itself.

    Attributes:
        field: Dot path of the value on each row.
        param: Query parameter that carries the value on the next run.
        template: Parameter text with a ``{value}`` placeholder; ``None`` sends
            the bare value.
        format: strftime pattern the value is rendered with, in UTC, before
            the template, for an API whose query language takes another time
            form than its records carry (``"%Y-%m-%d %H:%M"``). The stored
            value is read as ISO 8601 (``Z``, ``+00:00`` or ``+0000`` offsets,
            fractional seconds; no offset means UTC), or as epoch seconds when
            it is a number or a string of one.
            ``None`` renders it unchanged.
    """

    field: str = Field(..., min_length=1)
    param: str = Field(..., min_length=1)
    template: str | None = None
    format: str | None = None

    @field_validator("format")
    @classmethod
    def _check_format(cls, value: str | None) -> str | None:
        if value is None:
            return value
        directives = _STRFTIME_DIRECTIVE.findall(value)
        unknown = sorted({d for d in directives if d[1:] not in _STRFTIME_CODES})
        if unknown:
            raise ValueError(f"watermark format has unknown directives: {unknown}")
        if not [d for d in directives if d != "%%"]:
            raise ValueError(
                f"watermark format {value!r} has no strftime directive, such as %Y"
            )
        return value

    def format_value(self, value: Any) -> Any:
        """*value* as the API takes it: rendered by ``format``, else unchanged.

        Raises:
            ValueError: With ``format`` set, when *value* is neither ISO 8601
                nor epoch seconds.
        """
        if self.format is None:
            return value
        return _as_utc(value).strftime(self.format)

    def render(self, value: Any) -> dict[str, Any]:
        """Return the query parameters that resume from *value*."""
        value = self.format_value(value)
        if self.template is None:
            return {self.param: value}
        return {self.param: self.template.replace("{value}", str(value))}

Attributes¶

`field = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶

`format = None` `class-attribute` `instance-attribute` ¶

`param = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶

`template = None` `class-attribute` `instance-attribute` ¶

Methods:¶

`format_value(value)` ¶

value as the API takes it: rendered by format, else unchanged.

Raises:

Type	Description
`ValueError`	With `format` set, when value is neither ISO 8601 nor epoch seconds.

Source code in graflo/architecture/contract/bindings/connectors.py

def format_value(self, value: Any) -> Any:
    """*value* as the API takes it: rendered by ``format``, else unchanged.

    Raises:
        ValueError: With ``format`` set, when *value* is neither ISO 8601
            nor epoch seconds.
    """
    if self.format is None:
        return value
    return _as_utc(value).strftime(self.format)

`render(value)` ¶

Return the query parameters that resume from value.

Source code in graflo/architecture/contract/bindings/connectors.py

def render(self, value: Any) -> dict[str, Any]:
    """Return the query parameters that resume from *value*."""
    value = self.format_value(value)
    if self.template is None:
        return {self.param: value}
    return {self.param: self.template.replace("{value}", str(value))}

Functions:¶

`render_body(body, values)` ¶

Fill a request body: a string as text, a JSON structure by :func:render_param.

Source code in graflo/architecture/contract/bindings/connectors.py

def render_body(body: Any, values: Mapping[str, Any]) -> Any:
    """Fill a request body: a string as text, a JSON structure by :func:`render_param`."""
    if isinstance(body, str):
        return render_text(body, values)
    return render_param(body, values)

`render_param(value, values)` ¶

Fill the strings of a JSON-like value, at any depth; keys stay as written.

A string that is exactly {name} for a name in values becomes the value itself, its type kept; any other string is filled as text.

Source code in graflo/architecture/contract/bindings/connectors.py

def render_param(value: Any, values: Mapping[str, Any]) -> Any:
    """Fill the strings of a JSON-like *value*, at any depth; keys stay as written.

    A string that is exactly ``{name}`` for a name in *values* becomes the
    value itself, its type kept; any other string is filled as text.
    """
    if isinstance(value, str):
        whole = TEMPLATE_VARIABLE.fullmatch(value)
        if whole is not None and whole.group(1) in values:
            return values[whole.group(1)]
        return render_text(value, values)
    if isinstance(value, Mapping):
        return {key: render_param(item, values) for key, item in value.items()}
    if isinstance(value, list):
        return [render_param(item, values) for item in value]
    return value

`render_text(template, values, *, quote_values=False)` ¶

Fill {name} for every name in values; any other brace pair stays as written.

With quote_values each value is percent-encoded, for a URL path.

Source code in graflo/architecture/contract/bindings/connectors.py

def render_text(
    template: str, values: Mapping[str, Any], *, quote_values: bool = False
) -> str:
    """Fill ``{name}`` for every name in *values*; any other brace pair stays as written.

    With *quote_values* each value is percent-encoded, for a URL path.
    """

    def substitute(match: re.Match[str]) -> str:
        name = match.group(1)
        if name not in values:
            return match.group(0)
        text = str(values[name])
        return quote(text, safe="") if quote_values else text

    return TEMPLATE_VARIABLE.sub(substitute, template)

`template_variables(*parts)` ¶

The {name} variables in the strings of parts, at any depth.

Source code in graflo/architecture/contract/bindings/connectors.py

def template_variables(*parts: Any) -> set[str]:
    """The ``{name}`` variables in the strings of *parts*, at any depth."""
    found: set[str] = set()
    for part in parts:
        if isinstance(part, str):
            found.update(TEMPLATE_VARIABLE.findall(part))
        elif isinstance(part, Mapping):
            found |= template_variables(*part.values())
        elif isinstance(part, list):
            found |= template_variables(*part)
    return found

graflo.architecture.contract.bindings.connectors¶

Attributes¶

TEMPLATE_VARIABLE = re.compile('\\{([A-Za-z_][A-Za-z0-9_]*)\\}') module-attribute ¶

TemplateValue = str | int | float | bool module-attribute ¶

Classes¶

APIConnector ¶

Attributes¶

body = None class-attribute instance-attribute ¶

headers = Field(default_factory=dict) class-attribute instance-attribute ¶

method = 'GET' class-attribute instance-attribute ¶

pagination = None class-attribute instance-attribute ¶

params = Field(default_factory=dict) class-attribute instance-attribute ¶

parent = None class-attribute instance-attribute ¶

path = Field(..., description='Relative API endpoint path') class-attribute instance-attribute ¶

retries = 0 class-attribute instance-attribute ¶

retry_backoff_factor = 0.1 class-attribute instance-attribute ¶

retry_status_forcelist = Field(default_factory=lambda: [429, 500, 502, 503, 504]) class-attribute instance-attribute ¶

timeout = None class-attribute instance-attribute ¶

verify = True class-attribute instance-attribute ¶

watermark = None class-attribute instance-attribute ¶

Methods:¶

bound_source_kind() ¶

build_api_config(*, base_url, auth=None, default_headers=None, page_size_override=None) ¶

matches(resource_identifier) ¶

rendered(values) ¶

ApiResponseStructure ¶

Attributes¶

auto_detect = Field(default=False, description='When true, infer unset response paths from the first response body.') class-attribute instance-attribute ¶

batch_metadata_paths = Field(default_factory=dict, description='Map row annotation keys to response dot paths (e.g. ``_batch_id: result_id``).') class-attribute instance-attribute ¶

cursor_path = Field(default=None, description='Dot path to the next opaque cursor token.') class-attribute instance-attribute ¶

error_path = Field(default=None, description='Dot path to an error code in the response body (e.g. ``error``). Named in the raised error; a non-empty value raises on its own.') class-attribute instance-attribute ¶

has_more_path = Field(default=None, description='Dot path to a boolean more-pages flag (e.g. ``has_more``).') class-attribute instance-attribute ¶

next_offset_path = Field(default=None, description='Dot path to server-provided next offset for the following request (e.g. ``next_offset``).') class-attribute instance-attribute ¶

offset_path = Field(default=None, description='Dot path to echoed page start index (e.g. ``offset``).') class-attribute instance-attribute ¶

ok_path = Field(default=None, description='Dot path to a success flag in every response body (e.g. ``ok``). A body where it is not ``true`` raises, even with HTTP status 200.') class-attribute instance-attribute ¶

records_path = Field(default=None, description='Dot path to the record list (e.g. ``results``).') class-attribute instance-attribute ¶

total_count_path = Field(default=None, description='Dot path to total item count across all pages (e.g. ``count``).') class-attribute instance-attribute ¶

BoundSourceKind ¶

Attributes¶

API = 'api' class-attribute instance-attribute ¶

FILE = 'file' class-attribute instance-attribute ¶

KAFKA = 'kafka' class-attribute instance-attribute ¶

PUSH = 'push' class-attribute instance-attribute ¶

SPARQL = 'sparql' class-attribute instance-attribute ¶

SQL_TABLE = 'sql_table' class-attribute instance-attribute ¶

ConnectorUpdate ¶

Attributes¶

connector = Field(..., description='Connector reference: name or hash of the connector to patch.') class-attribute instance-attribute ¶

model_config = ConfigDict(extra='allow') class-attribute instance-attribute ¶

Methods:¶

as_patch() ¶

FileConnector ¶

Attributes¶

file_type = None class-attribute instance-attribute ¶

recursive = False class-attribute instance-attribute ¶

regex = None class-attribute instance-attribute ¶

sub_path = Field(default_factory=lambda: pathlib.Path('./')) class-attribute instance-attribute ¶

takes_default_conn_proxy = False class-attribute ¶

time_filter = None class-attribute instance-attribute ¶

Methods:¶

bound_source_kind() ¶

matches(resource_identifier) ¶

KafkaConnector ¶

Attributes¶

auto_offset_reset = 'earliest' class-attribute instance-attribute ¶

group_id = Field(..., description='Consumer group id') class-attribute instance-attribute ¶

idle_ms = Field(default=2000, ge=0, description='Stop after this many ms with no messages (0 disables idle stop).') class-attribute instance-attribute ¶

include_headers = False class-attribute instance-attribute ¶

max_wait_ms = Field(default=None, ge=0, description='Optional wall-clock cap for one consume run.') class-attribute instance-attribute ¶

poll_timeout_ms = Field(default=500, ge=1, description='Timeout for each consumer poll call.') class-attribute instance-attribute ¶

topics = Field(..., min_length=1, description='Kafka topics to consume') class-attribute instance-attribute ¶

value_encoding = 'json' class-attribute instance-attribute ¶

Methods:¶

bound_source_kind() ¶

build_kafka_config(*, conn) ¶

matches(resource_identifier) ¶

PaginationConfig ¶

Attributes¶

request = Field(default_factory=PaginationRequestConfig) class-attribute instance-attribute ¶

response = Field(default_factory=ApiResponseStructure) class-attribute instance-attribute ¶

`graflo.architecture.contract.bindings.connectors`¶

`TEMPLATE_VARIABLE = re.compile('\\{([A-Za-z_][A-Za-z0-9_]*)\\}')` `module-attribute` ¶

`TemplateValue = str | int | float | bool` `module-attribute` ¶

`APIConnector` ¶

`body = None` `class-attribute` `instance-attribute` ¶

`headers = Field(default_factory=dict)` `class-attribute` `instance-attribute` ¶

`method = 'GET'` `class-attribute` `instance-attribute` ¶

`pagination = None` `class-attribute` `instance-attribute` ¶

`params = Field(default_factory=dict)` `class-attribute` `instance-attribute` ¶

`parent = None` `class-attribute` `instance-attribute` ¶

`path = Field(..., description='Relative API endpoint path')` `class-attribute` `instance-attribute` ¶

`retries = 0` `class-attribute` `instance-attribute` ¶

`retry_backoff_factor = 0.1` `class-attribute` `instance-attribute` ¶

`retry_status_forcelist = Field(default_factory=lambda: [429, 500, 502, 503, 504])` `class-attribute` `instance-attribute` ¶

`timeout = None` `class-attribute` `instance-attribute` ¶

`verify = True` `class-attribute` `instance-attribute` ¶

`watermark = None` `class-attribute` `instance-attribute` ¶

`bound_source_kind()` ¶

`build_api_config(*, base_url, auth=None, default_headers=None, page_size_override=None)` ¶

`matches(resource_identifier)` ¶

`rendered(values)` ¶

`ApiResponseStructure` ¶

`auto_detect = Field(default=False, description='When true, infer unset response paths from the first response body.')` `class-attribute` `instance-attribute` ¶

batch_metadata_paths = Field(default_factory=dict, description='Map row annotation keys to response dot paths (e.g. ``_batch_id: result_id``).') `class-attribute` `instance-attribute` ¶

`cursor_path = Field(default=None, description='Dot path to the next opaque cursor token.')` `class-attribute` `instance-attribute` ¶

error_path = Field(default=None, description='Dot path to an error code in the response body (e.g. ``error``). Named in the raised error; a non-empty value raises on its own.') `class-attribute` `instance-attribute` ¶

has_more_path = Field(default=None, description='Dot path to a boolean more-pages flag (e.g. ``has_more``).') `class-attribute` `instance-attribute` ¶

next_offset_path = Field(default=None, description='Dot path to server-provided next offset for the following request (e.g. ``next_offset``).') `class-attribute` `instance-attribute` ¶

offset_path = Field(default=None, description='Dot path to echoed page start index (e.g. ``offset``).') `class-attribute` `instance-attribute` ¶

ok_path = Field(default=None, description='Dot path to a success flag in every response body (e.g. ``ok``). A body where it is not ``true`` raises, even with HTTP status 200.') `class-attribute` `instance-attribute` ¶

records_path = Field(default=None, description='Dot path to the record list (e.g. ``results``).') `class-attribute` `instance-attribute` ¶

total_count_path = Field(default=None, description='Dot path to total item count across all pages (e.g. ``count``).') `class-attribute` `instance-attribute` ¶

`BoundSourceKind` ¶

`API = 'api'` `class-attribute` `instance-attribute` ¶

`FILE = 'file'` `class-attribute` `instance-attribute` ¶

`KAFKA = 'kafka'` `class-attribute` `instance-attribute` ¶

`PUSH = 'push'` `class-attribute` `instance-attribute` ¶

`SPARQL = 'sparql'` `class-attribute` `instance-attribute` ¶

`SQL_TABLE = 'sql_table'` `class-attribute` `instance-attribute` ¶

`ConnectorUpdate` ¶

`connector = Field(..., description='Connector reference: name or hash of the connector to patch.')` `class-attribute` `instance-attribute` ¶

`model_config = ConfigDict(extra='allow')` `class-attribute` `instance-attribute` ¶

`as_patch()` ¶

`FileConnector` ¶

`file_type = None` `class-attribute` `instance-attribute` ¶

`recursive = False` `class-attribute` `instance-attribute` ¶

`regex = None` `class-attribute` `instance-attribute` ¶

`sub_path = Field(default_factory=lambda: pathlib.Path('./'))` `class-attribute` `instance-attribute` ¶

`takes_default_conn_proxy = False` `class-attribute` ¶

`time_filter = None` `class-attribute` `instance-attribute` ¶

`bound_source_kind()` ¶

`matches(resource_identifier)` ¶

`KafkaConnector` ¶

`auto_offset_reset = 'earliest'` `class-attribute` `instance-attribute` ¶

`group_id = Field(..., description='Consumer group id')` `class-attribute` `instance-attribute` ¶

`idle_ms = Field(default=2000, ge=0, description='Stop after this many ms with no messages (0 disables idle stop).')` `class-attribute` `instance-attribute` ¶

`include_headers = False` `class-attribute` `instance-attribute` ¶

`max_wait_ms = Field(default=None, ge=0, description='Optional wall-clock cap for one consume run.')` `class-attribute` `instance-attribute` ¶

`poll_timeout_ms = Field(default=500, ge=1, description='Timeout for each consumer poll call.')` `class-attribute` `instance-attribute` ¶

`topics = Field(..., min_length=1, description='Kafka topics to consume')` `class-attribute` `instance-attribute` ¶

`value_encoding = 'json'` `class-attribute` `instance-attribute` ¶

`bound_source_kind()` ¶

`build_kafka_config(*, conn)` ¶

`matches(resource_identifier)` ¶

`PaginationConfig` ¶

`request = Field(default_factory=PaginationRequestConfig)` `class-attribute` `instance-attribute` ¶

`response = Field(default_factory=ApiResponseStructure)` `class-attribute` `instance-attribute` ¶

`PaginationRequestConfig` ¶

`cursor_param = 'cursor'` `class-attribute` `instance-attribute` ¶

`initial_cursor = None` `class-attribute` `instance-attribute` ¶

`initial_offset = 0` `class-attribute` `instance-attribute` ¶

`initial_page = 1` `class-attribute` `instance-attribute` ¶

limit_param = Field(default='limit', description='Query parameter name for page size (offset strategy only). The value sent is ``page_size``, not a total item cap. When ``None``, no page-size query parameter is sent.') `class-attribute` `instance-attribute` ¶

`offset_param = 'offset'` `class-attribute` `instance-attribute` ¶

`page_param = 'page'` `class-attribute` `instance-attribute` ¶

page_size = Field(default=100, description='Records requested per HTTP page. Sent as the value of ``limit_param`` (offset) or ``per_page_param`` (page) when those param names are set.') `class-attribute` `instance-attribute` ¶

per_page_param = Field(default='per_page', description='Query parameter name for page size (page strategy only). The value sent is ``page_size``, not a total item cap.') `class-attribute` `instance-attribute` ¶

`strategy = 'offset'` `class-attribute` `instance-attribute` ¶

`ParentRequest` ¶

`connector = Field(..., min_length=1)` `class-attribute` `instance-attribute` ¶