graflo.architecture.contract

Declarative contracts: manifest, bindings, ingestion models, resources, transforms.

Modules:

Name	Description
bindings	Resource connectors and named binding collections.
ingestion	Declarative ingestion contract: resources, transforms, steps, and ingestion model.
manifest	Graph manifest model for complete ingestion contracts.
provenance	Manifest-level metadata, and the provenance stamping that writes part of it.

Attributes

Bindings = BindingsRegistry module-attribute

__all__ = ['APIConnector', 'ApiResponseStructure', 'Bindings', 'BoundSourceKind', 'ColumnTimeFilter', 'FileConnector', 'GraphManifest', 'IngestionModel', 'KafkaConnector', 'PaginationConfig', 'PaginationRequestConfig', 'ProtoTransform', 'Resource', 'ResourceConfig', 'ResourceConnector', 'ResourceRuntime', 'SparqlConnector', 'TableConnector', 'Transform', 'build_resource_runtime'] module-attribute

Classes

APIConnector

Bases: ResourceConnector

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.provider.ApiGeneralizedConnConfig.

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

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

class APIConnector(ResourceConnector):
    """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.provider.ApiGeneralizedConnConfig`.

    Attributes:
        path: Relative endpoint path (e.g. ``/api/users``).
        method: HTTP method (default ``GET``).
        params: Static query parameters.
        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.
    """

    path: str = Field(..., description="Relative API endpoint path")
    method: str = "GET"
    params: dict[str, Any] = Field(default_factory=dict)
    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

    @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),
            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),
        )

Attributes

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

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

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),
        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),
    )

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}

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.",
    )
    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

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

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

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).
    """

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

Attributes

API = 'api' class-attribute instance-attribute

FILE = 'file' class-attribute instance-attribute

KAFKA = 'kafka' class-attribute instance-attribute

SPARQL = 'sparql' class-attribute instance-attribute

SQL_TABLE = 'sql_table' class-attribute instance-attribute

ColumnTimeFilter

Bases: ConfigBaseModel

Predicate on a single date/time column (SQL-friendly via :class:FilterExpression).

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

class ColumnTimeFilter(ConfigBaseModel):
    """Predicate on a single date/time column (SQL-friendly via :class:`FilterExpression`)."""

    column: str = Field(..., description="Column name for the time predicate.")
    start: str | None = Field(
        default=None,
        description="Lower bound (ISO date or datetime). Interpreted with start_inclusive.",
    )
    end: str | None = Field(
        default=None,
        description="Upper bound (ISO date or datetime). Interpreted with end_inclusive.",
    )
    interval: str | None = Field(
        default=None,
        description='Pandas timedelta string (e.g. "7D", "2H"); requires start; '
        "defines [start, start + interval). Mutually exclusive with end.",
    )
    not_equals: str | None = Field(
        default=None,
        description="If set, render column != value. Mutually exclusive with start/end/interval.",
    )
    start_inclusive: bool = Field(
        default=True,
        description="If True (default), lower bound uses >= when start is set.",
    )
    end_inclusive: bool = Field(
        default=False,
        description="If False (default), upper bound uses < when end is set.",
    )

    @model_validator(mode="after")
    def _check_shape(self) -> Self:
        has_pred = any(
            v is not None
            for v in (self.start, self.end, self.interval, self.not_equals)
        )
        if not has_pred:
            # Column-only hint (e.g. introspection sets datetime column without a WHERE).
            return self
        if self.not_equals is not None:
            if any(v is not None for v in (self.start, self.end, self.interval)):
                raise ValueError(
                    "not_equals cannot be combined with start, end, or interval"
                )
            return self
        if self.interval is not None:
            if self.start is None:
                raise ValueError("interval requires start")
            if self.end is not None:
                raise ValueError(
                    "interval cannot be combined with end; use start + interval"
                )
            self._validated_timedelta()
            return self
        if self.start is None and self.end is None:
            raise ValueError(
                "ColumnTimeFilter requires at least one of: start, end, interval, not_equals"
            )
        return self

    def _validated_timedelta(self) -> pd.Timedelta:
        assert self.interval is not None
        try:
            td = pd.Timedelta(self.interval)
        except (ValueError, TypeError) as e:
            raise ValueError(
                f"Invalid pandas timedelta string for interval: {self.interval!r}"
            ) from e
        if pd.isna(td):
            raise ValueError(
                f"Invalid pandas timedelta string for interval: {self.interval!r}"
            )
        return td

    def _interval_upper_literal(self) -> str:
        assert self.start is not None and self.interval is not None
        start_dt, date_only = parse_iso_date_or_datetime(self.start)
        delta = self._validated_timedelta().to_pytimedelta()
        end_dt = start_dt + delta
        upper_date_only = date_only and (end_dt.time() == time.min)
        return format_sql_literal(end_dt, upper_date_only)

    def _lower_literal(self) -> str:
        assert self.start is not None
        dt, date_only = parse_iso_date_or_datetime(self.start)
        return format_sql_literal(dt, date_only)

    def _upper_literal(self) -> str:
        assert self.end is not None
        dt, date_only = parse_iso_date_or_datetime(self.end)
        return format_sql_literal(dt, date_only)

    def as_filter_expression(self) -> FilterExpression | None:
        """Return a single composite AND of leaves, or None if misconfigured."""
        from graflo.filter.onto import (
            ComparisonOperator,
            FilterExpression,
            LogicalOperator,
        )

        if self.not_equals is not None:
            return FilterExpression(
                kind="leaf",
                field=self.column,
                cmp_operator=ComparisonOperator.NEQ,
                value=[self.not_equals],
            )

        leaves: list[FilterExpression] = []

        if self.interval is not None:
            assert self.start is not None
            # Half-open window [start, start + interval).
            leaves.append(
                FilterExpression(
                    kind="leaf",
                    field=self.column,
                    cmp_operator=ComparisonOperator.GE,
                    value=[self._lower_literal()],
                )
            )
            leaves.append(
                FilterExpression(
                    kind="leaf",
                    field=self.column,
                    cmp_operator=ComparisonOperator.LT,
                    value=[self._interval_upper_literal()],
                )
            )
        else:
            if self.start is not None:
                lower_op = (
                    ComparisonOperator.GE
                    if self.start_inclusive
                    else ComparisonOperator.GT
                )
                leaves.append(
                    FilterExpression(
                        kind="leaf",
                        field=self.column,
                        cmp_operator=lower_op,
                        value=[self._lower_literal()],
                    )
                )
            if self.end is not None:
                upper_op = (
                    ComparisonOperator.LE
                    if self.end_inclusive
                    else ComparisonOperator.LT
                )
                leaves.append(
                    FilterExpression(
                        kind="leaf",
                        field=self.column,
                        cmp_operator=upper_op,
                        value=[self._upper_literal()],
                    )
                )

        if not leaves:
            return None
        if len(leaves) == 1:
            return leaves[0]
        return FilterExpression(
            kind="composite",
            operator=LogicalOperator.AND,
            deps=leaves,
        )

Attributes

column = Field(..., description='Column name for the time predicate.') class-attribute instance-attribute

end = Field(default=None, description='Upper bound (ISO date or datetime). Interpreted with end_inclusive.') class-attribute instance-attribute

end_inclusive = Field(default=False, description='If False (default), upper bound uses < when end is set.') class-attribute instance-attribute

interval = Field(default=None, description='Pandas timedelta string (e.g. "7D", "2H"); requires start; defines [start, start + interval). Mutually exclusive with end.') class-attribute instance-attribute

not_equals = Field(default=None, description='If set, render column != value. Mutually exclusive with start/end/interval.') class-attribute instance-attribute

start = Field(default=None, description='Lower bound (ISO date or datetime). Interpreted with start_inclusive.') class-attribute instance-attribute

start_inclusive = Field(default=True, description='If True (default), lower bound uses >= when start is set.') class-attribute instance-attribute

Methods:

as_filter_expression()

Return a single composite AND of leaves, or None if misconfigured.

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

def as_filter_expression(self) -> FilterExpression | None:
    """Return a single composite AND of leaves, or None if misconfigured."""
    from graflo.filter.onto import (
        ComparisonOperator,
        FilterExpression,
        LogicalOperator,
    )

    if self.not_equals is not None:
        return FilterExpression(
            kind="leaf",
            field=self.column,
            cmp_operator=ComparisonOperator.NEQ,
            value=[self.not_equals],
        )

    leaves: list[FilterExpression] = []

    if self.interval is not None:
        assert self.start is not None
        # Half-open window [start, start + interval).
        leaves.append(
            FilterExpression(
                kind="leaf",
                field=self.column,
                cmp_operator=ComparisonOperator.GE,
                value=[self._lower_literal()],
            )
        )
        leaves.append(
            FilterExpression(
                kind="leaf",
                field=self.column,
                cmp_operator=ComparisonOperator.LT,
                value=[self._interval_upper_literal()],
            )
        )
    else:
        if self.start is not None:
            lower_op = (
                ComparisonOperator.GE
                if self.start_inclusive
                else ComparisonOperator.GT
            )
            leaves.append(
                FilterExpression(
                    kind="leaf",
                    field=self.column,
                    cmp_operator=lower_op,
                    value=[self._lower_literal()],
                )
            )
        if self.end is not None:
            upper_op = (
                ComparisonOperator.LE
                if self.end_inclusive
                else ComparisonOperator.LT
            )
            leaves.append(
                FilterExpression(
                    kind="leaf",
                    field=self.column,
                    cmp_operator=upper_op,
                    value=[self._upper_literal()],
                )
            )

    if not leaves:
        return None
    if len(leaves) == 1:
        return leaves[0]
    return FilterExpression(
        kind="composite",
        operator=LogicalOperator.AND,
        deps=leaves,
    )

FileConnector

Bases: ResourceConnector

Connector for matching files.

Attributes:

Name	Type	Description
regex	str | None	Regular expression pattern for matching filenames
sub_path	Path	Path to search for matching files (default: "./")
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: Path to search for matching files (default: "./")
        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`.
    """

    regex: str | None = None
    sub_path: pathlib.Path = Field(default_factory=lambda: pathlib.Path("./"))
    time_filter: ColumnTimeFilter | None = None

    @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

regex = None class-attribute instance-attribute

sub_path = Field(default_factory=lambda: pathlib.Path('./')) class-attribute instance-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))

GraphManifest

Bases: ConfigBaseModel

Canonical config contract for graph schema, ingestion, and bindings.

Source code in graflo/architecture/contract/manifest.py

class GraphManifest(ConfigBaseModel):
    """Canonical config contract for graph schema, ingestion, and bindings."""

    model_config = ConfigDict(populate_by_name=True)

    graph_schema: Schema | None = PydanticField(
        default=None,
        description="Logical graph schema contract.",
        validation_alias=AliasChoices("schema", "graph_schema"),
        serialization_alias="schema",
    )
    ingestion_model: IngestionModel | None = PydanticField(
        default=None,
        description="Ingestion resources and transforms.",
    )
    bindings: Bindings | None = PydanticField(
        default=None,
        description="Bindings mapping resources to concrete data sources.",
    )
    metadata: ManifestMetadata | None = PydanticField(
        default=None,
        description=(
            "Manifest-level identity: name, description and provenance. "
            "Excluded from this manifest's own content hash — a content "
            "address that covered its name or its lineage would not be "
            "path-independent, and two routes to the same world model could "
            "never be recognised as equal."
        ),
    )

    @classmethod
    def from_config(cls, data: dict[str, Any]) -> GraphManifest:
        """Build a manifest from a Python mapping payload."""
        return cls.from_dict(data)

    @model_validator(mode="after")
    def _validate_manifest(self) -> GraphManifest:
        if (
            self.graph_schema is None
            and self.ingestion_model is None
            and self.bindings is None
        ):
            raise ValueError(
                "GraphManifest requires at least one block: "
                "schema, ingestion_model, or bindings."
            )
        return self

    def finish_init(
        self,
        *,
        strict_references: bool = False,
        dynamic_edge_feedback: bool = False,
    ) -> None:
        """Initialize model internals and cross-block runtime links."""
        if self.graph_schema is not None:
            self.graph_schema.finish_init()
        if self.graph_schema is not None and self.ingestion_model is not None:
            self.ingestion_model.finish_init(
                self.graph_schema.core_schema,
                strict_references=strict_references,
                dynamic_edge_feedback=dynamic_edge_feedback,
                target_db_flavor=self.graph_schema.db_profile.db_flavor,
            )

    def require_schema(self) -> Schema:
        if self.graph_schema is None:
            raise ValueError("GraphManifest is missing required 'schema' block.")
        return self.graph_schema

    def require_ingestion_model(self) -> IngestionModel:
        if self.ingestion_model is None:
            raise ValueError(
                "GraphManifest is missing required 'ingestion_model' block."
            )
        return self.ingestion_model

    def require_bindings(self) -> Bindings:
        if self.bindings is None:
            raise ValueError("GraphManifest is missing required 'bindings' block.")
        return self.bindings

Attributes

bindings = PydanticField(default=None, description='Bindings mapping resources to concrete data sources.') class-attribute instance-attribute

graph_schema = PydanticField(default=None, description='Logical graph schema contract.', validation_alias=AliasChoices('schema', 'graph_schema'), serialization_alias='schema') class-attribute instance-attribute

ingestion_model = PydanticField(default=None, description='Ingestion resources and transforms.') class-attribute instance-attribute

metadata = PydanticField(default=None, description="Manifest-level identity: name, description and provenance. Excluded from this manifest's own content hash — a content address that covered its name or its lineage would not be path-independent, and two routes to the same world model could never be recognised as equal.") class-attribute instance-attribute

model_config = ConfigDict(populate_by_name=True) class-attribute instance-attribute

Methods:

finish_init(*, strict_references=False, dynamic_edge_feedback=False)

Initialize model internals and cross-block runtime links.

Source code in graflo/architecture/contract/manifest.py

def finish_init(
    self,
    *,
    strict_references: bool = False,
    dynamic_edge_feedback: bool = False,
) -> None:
    """Initialize model internals and cross-block runtime links."""
    if self.graph_schema is not None:
        self.graph_schema.finish_init()
    if self.graph_schema is not None and self.ingestion_model is not None:
        self.ingestion_model.finish_init(
            self.graph_schema.core_schema,
            strict_references=strict_references,
            dynamic_edge_feedback=dynamic_edge_feedback,
            target_db_flavor=self.graph_schema.db_profile.db_flavor,
        )

from_config(data) classmethod

Build a manifest from a Python mapping payload.

Source code in graflo/architecture/contract/manifest.py

@classmethod
def from_config(cls, data: dict[str, Any]) -> GraphManifest:
    """Build a manifest from a Python mapping payload."""
    return cls.from_dict(data)

require_bindings()

Source code in graflo/architecture/contract/manifest.py

def require_bindings(self) -> Bindings:
    if self.bindings is None:
        raise ValueError("GraphManifest is missing required 'bindings' block.")
    return self.bindings

require_ingestion_model()

Source code in graflo/architecture/contract/manifest.py

def require_ingestion_model(self) -> IngestionModel:
    if self.ingestion_model is None:
        raise ValueError(
            "GraphManifest is missing required 'ingestion_model' block."
        )
    return self.ingestion_model

require_schema()

Source code in graflo/architecture/contract/manifest.py

def require_schema(self) -> Schema:
    if self.graph_schema is None:
        raise ValueError("GraphManifest is missing required 'schema' block.")
    return self.graph_schema

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.provider.KafkaGeneralizedConnConfig.

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

    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

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.",
    )
    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."""

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

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
    """

SparqlConnector

Bases: ResourceConnector

Connector for matching SPARQL / RDF data sources.

Each SparqlConnector targets instances of a single rdf:Class. 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").
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(ResourceConnector):
    """Connector for matching SPARQL / RDF data sources.

    Each ``SparqlConnector`` targets instances of a single ``rdf:Class``.
    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"``).
        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 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

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

    @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

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

Functions:

__getattr__(name)

Source code in graflo/architecture/contract/__init__.py

def __getattr__(name: str) -> Any:
    # Lazy: ingestion → runtime → graph_types; keep bindings importable without that cycle.
    if name in _INGESTION_EXPORTS:
        from . import ingestion as _ingestion

        value = getattr(_ingestion, name)
        globals()[name] = value
        return value
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")