graflo.hq.ingestion_parameters

Ingestion parameters and per-document cast-failure models for the caster.

This module exists to keep graflo/hq/caster.py focused on casting logic, while keeping ingestion-policy types stable and importable.

Classes

CastBatchResult

Bases: BaseModel

Outcome of casting a batch through a resource (possibly with skipped documents).

Source code in graflo/hq/ingestion_parameters.py

class CastBatchResult(BaseModel):
    """Outcome of casting a batch through a resource (possibly with skipped documents)."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    graph: GraphContainer
    failures: list[DocCastFailure] = Field(default_factory=list)

Attributes

failures = Field(default_factory=list) class-attribute instance-attribute

graph instance-attribute

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

DocCastFailure

Bases: BaseModel

Structured record for one source document that failed during resource casting.

Source code in graflo/hq/ingestion_parameters.py

class DocCastFailure(BaseModel):
    """Structured record for one source document that failed during resource casting."""

    resource_name: str
    doc_index: int
    exception_type: str
    message: str
    failure_kind: Literal["document", "transform"] = "document"
    traceback: str = Field(
        default="",
        description="Formatted traceback, truncated to the configured max length.",
    )
    doc_preview: Any = Field(
        default=None,
        description="Subset or truncated JSON of the source document for debugging.",
    )
    location_path: tuple[str | int | None, ...] | None = Field(
        default=None,
        description="Extraction location path when failure_kind is transform.",
    )
    transform_label: str | None = Field(
        default=None,
        description="Transform name or module.foo when failure_kind is transform.",
    )
    nulled_fields: tuple[str, ...] | None = Field(
        default=None,
        description="Output fields set to None when failure_kind is transform.",
    )

Attributes

doc_index instance-attribute

doc_preview = Field(default=None, description='Subset or truncated JSON of the source document for debugging.') class-attribute instance-attribute

exception_type instance-attribute

failure_kind = 'document' class-attribute instance-attribute

location_path = Field(default=None, description='Extraction location path when failure_kind is transform.') class-attribute instance-attribute

message instance-attribute

nulled_fields = Field(default=None, description='Output fields set to None when failure_kind is transform.') class-attribute instance-attribute

resource_name instance-attribute

traceback = Field(default='', description='Formatted traceback, truncated to the configured max length.') class-attribute instance-attribute

transform_label = Field(default=None, description='Transform name or module.foo when failure_kind is transform.') class-attribute instance-attribute

DocErrorBudgetExceeded

Bases: RuntimeError

Raised when total document cast failures exceed IngestionParams.max_doc_errors.

Source code in graflo/hq/ingestion_parameters.py

class DocErrorBudgetExceeded(RuntimeError):
    """Raised when total document cast failures exceed ``IngestionParams.max_doc_errors``."""

    def __init__(
        self,
        *,
        total_failures: int,
        limit: int,
        doc_error_sink_path: Path | None,
    ) -> None:
        self.total_failures = total_failures
        self.limit = limit
        self.doc_error_sink_path = doc_error_sink_path
        sink = str(doc_error_sink_path) if doc_error_sink_path else "(not configured)"
        super().__init__(
            f"Document error budget exceeded: {total_failures} total failures "
            f"(limit {limit}). Doc error sink (jsonl.gz): {sink}"
        )

Attributes

doc_error_sink_path = doc_error_sink_path instance-attribute

limit = limit instance-attribute

total_failures = total_failures instance-attribute

Methods:

__init__(*, total_failures, limit, doc_error_sink_path)

Source code in graflo/hq/ingestion_parameters.py

def __init__(
    self,
    *,
    total_failures: int,
    limit: int,
    doc_error_sink_path: Path | None,
) -> None:
    self.total_failures = total_failures
    self.limit = limit
    self.doc_error_sink_path = doc_error_sink_path
    sink = str(doc_error_sink_path) if doc_error_sink_path else "(not configured)"
    super().__init__(
        f"Document error budget exceeded: {total_failures} total failures "
        f"(limit {limit}). Doc error sink (jsonl.gz): {sink}"
    )

IngestionParams

Bases: BaseModel

Parameters for controlling the ingestion process.

max_items caps how many source items (rows, JSON objects, grouped RDF subjects, …) are read per resource run. It maps to AbstractDataSource.iter_batches(..., limit=...). batch_size is only the maximum number of items per yielded batch, not a cap on total volume.

Source code in graflo/hq/ingestion_parameters.py

class IngestionParams(BaseModel):
    """Parameters for controlling the ingestion process.

    ``max_items`` caps how many **source items** (rows, JSON objects, grouped
    RDF subjects, …) are read per resource run. It maps to
    ``AbstractDataSource.iter_batches(..., limit=...)``. ``batch_size`` is only
    the maximum number of items per yielded batch, not a cap on total volume.
    """

    clear_data: bool = False
    n_cores: int = Field(
        default=1,
        ge=1,
        description=(
            "Number of cast worker processes. With cast_executor='auto' (default), "
            "n_cores > 1 routes large batches to worker processes (workers fold and "
            "filter their own chunks, so the parallelism is real end to end); "
            "1 keeps casting in-process."
        ),
    )
    cast_executor: Literal["auto", "inline", "thread", "process"] = Field(
        default="auto",
        description=(
            "How documents are cast. Leave on 'auto': with n_cores=1 casting "
            "runs in-process; with n_cores > 1 large batches are spread over "
            "worker processes automatically. 'inline' pins casting in-process "
            "regardless of n_cores; 'process' always uses worker processes; "
            "'thread' is a legacy escape hatch (GIL-bound, rarely useful). "
            "With dynamic_edges=True this setting is effectively ignored — "
            "edge discovery is order-dependent, so casting always runs "
            "sequentially in-process; no action needed on your side. See the "
            "'Parallelism' concept page for the full picture."
        ),
    )
    max_concurrent_sources: int | None = Field(
        default=None,
        ge=1,
        description=(
            "How many data sources of one resource are processed concurrently "
            "(sources of a resource are independent shards, e.g. one file "
            "each). Defaults to min(4, sources in the resource). Resources "
            "themselves always run in declaration order — later resources may "
            "depend on earlier ones' database state."
        ),
    )
    max_items: int | None = Field(
        default=None,
        ge=1,
        description=(
            "Maximum number of source items (rows / JSON objects / grouped "
            "RDF subjects) to ingest for each resource. Not a batch count."
        ),
    )
    batch_size: int = Field(
        default=10000,
        ge=1,
        description="Number of source items to group per batch for casting and writes.",
    )
    batch_prefetch: int = Field(
        default=2,
        ge=1,
        description=(
            "How many batches to prefetch ahead while processing current batch. "
            "Keeps ingestion lazy with bounded memory."
        ),
    )
    max_in_flight_batches: int = Field(
        default=2,
        ge=1,
        description=(
            "How many batches of one data source may be cast/written concurrently, "
            "so casting batch N+1 overlaps writing batch N. Configurations where "
            "batch order is semantic are forced to 1 automatically and logged at "
            "INFO — see the 'Parallelism' concept page for the full list "
            "(dynamic_edges, blank vertices, extra_weights, secondary-identity "
            "endpoints, native bulk load, graflo_backend target). Set to 1 to "
            "disable overlap entirely."
        ),
    )
    dry: bool = False
    init_only: bool = False
    limit_files: int | None = None
    resources: list[str] | None = None
    connectors: list[str] | None = Field(
        default=None,
        description=(
            "Optional subset of connectors to ingest, by connector name or hash "
            "(same refs as bindings.resource_connector.connector). When set, only "
            "matching connectors are registered as data sources. Intersects with "
            "resources when both are set."
        ),
    )
    vertices: list[str] | None = None
    max_concurrent_db_ops: int = Field(
        default=8,
        ge=1,
        description=(
            "Upper bound on concurrent DB operations per batch. Writes are I/O-bound, "
            "so this is where concurrency actually pays; it used to default to "
            "n_cores, which meant writes were serial out of the box."
        ),
    )
    datetime_after: str | None = None
    datetime_before: str | None = None
    datetime_column: str | None = None

    # Strict contract checks for major-release style validation workflows.
    strict_references: bool = True
    strict_registry: bool = True
    dynamic_edges: bool = Field(
        default=False,
        description=(
            "Discover edges from the data during casting: a document may "
            "register an edge type the schema does not declare, and later "
            "documents can then infer over it. Because that feedback is "
            "order-dependent, the resource runs fully serial — casting stays "
            "in-process single-threaded, and batches and sources are processed "
            "one at a time (automatic; logged at INFO). Per-batch DB writes and "
            "batch prefetch stay concurrent. For throughput, use dynamic_edges "
            "as a discovery pass on a sample, add the discovered edges to the "
            "schema, then re-ingest with dynamic_edges off and full parallelism."
        ),
    )
    on_doc_error: Literal["skip", "fail"] = "skip"
    doc_error_sink_path: Path | None = Field(
        default=None,
        description=(
            "Append gzip-compressed JSONL cast-failure records (typical suffix .jsonl.gz)."
        ),
    )
    max_doc_errors: int | None = None
    doc_error_preview_max_bytes: int = 4096
    doc_error_preview_keys: tuple[str, ...] | None = None
    drop_empty_identity_docs: bool = Field(
        default=True,
        description=(
            "After casting, remove vertex docs and edge tuples where all schema "
            "identity fields for that vertex type are missing, null, or empty string."
        ),
    )

Attributes

batch_prefetch = Field(default=2, ge=1, description='How many batches to prefetch ahead while processing current batch. Keeps ingestion lazy with bounded memory.') class-attribute instance-attribute

batch_size = Field(default=10000, ge=1, description='Number of source items to group per batch for casting and writes.') class-attribute instance-attribute

cast_executor = Field(default='auto', description="How documents are cast. Leave on 'auto': with n_cores=1 casting runs in-process; with n_cores > 1 large batches are spread over worker processes automatically. 'inline' pins casting in-process regardless of n_cores; 'process' always uses worker processes; 'thread' is a legacy escape hatch (GIL-bound, rarely useful). With dynamic_edges=True this setting is effectively ignored — edge discovery is order-dependent, so casting always runs sequentially in-process; no action needed on your side. See the 'Parallelism' concept page for the full picture.") class-attribute instance-attribute

clear_data = False class-attribute instance-attribute

connectors = Field(default=None, description='Optional subset of connectors to ingest, by connector name or hash (same refs as bindings.resource_connector.connector). When set, only matching connectors are registered as data sources. Intersects with resources when both are set.') class-attribute instance-attribute

datetime_after = None class-attribute instance-attribute

datetime_before = None class-attribute instance-attribute

datetime_column = None class-attribute instance-attribute

doc_error_preview_keys = None class-attribute instance-attribute

doc_error_preview_max_bytes = 4096 class-attribute instance-attribute

doc_error_sink_path = Field(default=None, description='Append gzip-compressed JSONL cast-failure records (typical suffix .jsonl.gz).') class-attribute instance-attribute

drop_empty_identity_docs = Field(default=True, description='After casting, remove vertex docs and edge tuples where all schema identity fields for that vertex type are missing, null, or empty string.') class-attribute instance-attribute

dry = False class-attribute instance-attribute

dynamic_edges = Field(default=False, description='Discover edges from the data during casting: a document may register an edge type the schema does not declare, and later documents can then infer over it. Because that feedback is order-dependent, the resource runs fully serial — casting stays in-process single-threaded, and batches and sources are processed one at a time (automatic; logged at INFO). Per-batch DB writes and batch prefetch stay concurrent. For throughput, use dynamic_edges as a discovery pass on a sample, add the discovered edges to the schema, then re-ingest with dynamic_edges off and full parallelism.') class-attribute instance-attribute

init_only = False class-attribute instance-attribute

limit_files = None class-attribute instance-attribute

max_concurrent_db_ops = Field(default=8, ge=1, description='Upper bound on concurrent DB operations per batch. Writes are I/O-bound, so this is where concurrency actually pays; it used to default to n_cores, which meant writes were serial out of the box.') class-attribute instance-attribute

max_concurrent_sources = Field(default=None, ge=1, description="How many data sources of one resource are processed concurrently (sources of a resource are independent shards, e.g. one file each). Defaults to min(4, sources in the resource). Resources themselves always run in declaration order — later resources may depend on earlier ones' database state.") class-attribute instance-attribute

max_doc_errors = None class-attribute instance-attribute

max_in_flight_batches = Field(default=2, ge=1, description="How many batches of one data source may be cast/written concurrently, so casting batch N+1 overlaps writing batch N. Configurations where batch order is semantic are forced to 1 automatically and logged at INFO — see the 'Parallelism' concept page for the full list (dynamic_edges, blank vertices, extra_weights, secondary-identity endpoints, native bulk load, graflo_backend target). Set to 1 to disable overlap entirely.") class-attribute instance-attribute

max_items = Field(default=None, ge=1, description='Maximum number of source items (rows / JSON objects / grouped RDF subjects) to ingest for each resource. Not a batch count.') class-attribute instance-attribute

n_cores = Field(default=1, ge=1, description="Number of cast worker processes. With cast_executor='auto' (default), n_cores > 1 routes large batches to worker processes (workers fold and filter their own chunks, so the parallelism is real end to end); 1 keeps casting in-process.") class-attribute instance-attribute

on_doc_error = 'skip' class-attribute instance-attribute

resources = None class-attribute instance-attribute

strict_references = True class-attribute instance-attribute

strict_registry = True class-attribute instance-attribute

vertices = None class-attribute instance-attribute