graflo.hq.container_write

Write a graph container to a database by identity upsert, without a resource.

For producers that assemble a :class:GraphContainer themselves rather than casting rows through an ingestion resource. The schema is defined on the target first, and an existing definition is reused. Edges whose endpoint is known only by a secondary identity (:class:PendingEdge) are looked up in the target after the container is written, so an endpoint the container itself carries is found. Everything is upserted by identity through :class:DBWriter, so writing the same container twice leaves the database as it was after the first write. The file backend is the exception: it appends.

Attributes

Classes

ContainerWriteReport

Bases: ConfigBaseModel

What :func:write_container sent to the database.

Source code in graflo/hq/container_write.py

class ContainerWriteReport(ConfigBaseModel):
    """What :func:`write_container` sent to the database."""

    target: str = Field(description="Connection type written to.")
    schema_defined: bool = Field(
        description="False when an existing definition was reused."
    )
    vertices: dict[str, int] = Field(description="Documents per vertex type.")
    edges: dict[str, int] = Field(
        description="Edges per edge, keyed ``source-relation-target``."
    )
    lookups_resolved: int = Field(
        default=0, description="Pending edges whose endpoints were found."
    )
    lookups_unresolved: int = Field(
        default=0, description="Pending edges dropped: no match, or an ambiguous one."
    )

Attributes

edges = Field(description='Edges per edge, keyed ``source-relation-target``.') class-attribute instance-attribute

lookups_resolved = Field(default=0, description='Pending edges whose endpoints were found.') class-attribute instance-attribute

lookups_unresolved = Field(default=0, description='Pending edges dropped: no match, or an ambiguous one.') class-attribute instance-attribute

schema_defined = Field(description='False when an existing definition was reused.') class-attribute instance-attribute

target = Field(description='Connection type written to.') class-attribute instance-attribute

vertices = Field(description='Documents per vertex type.') class-attribute instance-attribute

Functions:

define_schema(schema, conn_conf, recreate)

Define schema on the target, or reuse its definition; True when defined.

Source code in graflo/hq/container_write.py

def define_schema(schema: Schema, conn_conf: DBConfig, recreate: bool) -> bool:
    """Define *schema* on the target, or reuse its definition; True when defined."""
    manifest = GraphManifest(
        graph_schema=schema, ingestion_model=IngestionModel(resources=[])
    )
    try:
        GraphEngine(target_db_flavor=conn_conf.connection_type).define_schema(
            manifest=manifest, target_db_config=conn_conf, recreate_schema=recreate
        )
    except SchemaExistsError:
        schema.db_profile.db_flavor = conn_conf.connection_type
        schema.finish_init()
        return False
    return True

edge_label(edge_id)

source-relation-target, the key of :attr:ContainerWriteReport.edges.

Source code in graflo/hq/container_write.py

def edge_label(edge_id: EdgeId) -> str:
    """``source-relation-target``, the key of :attr:`ContainerWriteReport.edges`."""
    return f"{edge_id[0]}-{edge_id[2] or ''}-{edge_id[1]}"

write_container(schema, container, conn_conf, *, pending=(), recreate_schema=False, ambiguity='skip', stamp=None, guard=None) async

Define schema on the target and upsert container by identity.

Parameters:

Name	Type	Description	Default
schema	Schema	The schema every document conforms to. Never mutated.	required
container	GraphContainer	Vertices and edges keyed by primary identity. Never mutated.	required
conn_conf	DBConfig	The target database.	required
pending	Sequence[PendingEdge]	Edges with an endpoint known only by a secondary identity; each is looked up in the target once container is written, and written when found.	()
recreate_schema	bool	Drop and redefine the schema instead of reusing it.	False
ambiguity	EndpointAmbiguityPolicy	What a secondary identity matching several vertices does. skip (the default) drops the edge rather than guess.	'skip'
stamp	Mapping[str, Any] | None	Values of the schema's reserved fields, set on everything written, pending edges included (DBWriter.write). Required when the container holds a type with a reserved field.	None
guard	FilterExpression | None	A trusted read guard every pending lookup must pass, so an endpoint is never resolved to a vertex the guard hides; refused, before anything is written, by a backend without read guards.	None

Returns:

Name	Type	Description
ContainerWriteReport	ContainerWriteReport	counts of what was sent and what was looked up.

Source code in graflo/hq/container_write.py

async def write_container(
    schema: Schema,
    container: GraphContainer,
    conn_conf: DBConfig,
    *,
    pending: Sequence[PendingEdge] = (),
    recreate_schema: bool = False,
    ambiguity: EndpointAmbiguityPolicy = "skip",
    stamp: Mapping[str, Any] | None = None,
    guard: FilterExpression | None = None,
) -> ContainerWriteReport:
    """Define *schema* on the target and upsert *container* by identity.

    Args:
        schema: The schema every document conforms to. Never mutated.
        container: Vertices and edges keyed by primary identity. Never mutated.
        conn_conf: The target database.
        pending: Edges with an endpoint known only by a secondary identity;
            each is looked up in the target once *container* is written, and
            written when found.
        recreate_schema: Drop and redefine the schema instead of reusing it.
        ambiguity: What a secondary identity matching several vertices does.
            ``skip`` (the default) drops the edge rather than guess.
        stamp: Values of the schema's reserved fields, set on everything
            written, pending edges included (``DBWriter.write``). Required when
            the container holds a type with a reserved field.
        guard: A trusted read guard every *pending* lookup must pass, so an
            endpoint is never resolved to a vertex the guard hides; refused,
            before anything is written, by a backend without read guards.

    Returns:
        ContainerWriteReport: counts of what was sent and what was looked up.
    """
    schema = schema.model_copy(deep=True)
    defined = await asyncio.to_thread(define_schema, schema, conn_conf, recreate_schema)
    to_write = GraphContainer(
        vertices={
            name: [dict(doc) for doc in docs]
            for name, docs in container.vertices.items()
        },
        edges={
            edge_id: [_copy_row(row) for row in triples]
            for edge_id, triples in container.edges.items()
        },
    )

    ingestion_model = IngestionModel(resources=[])
    ingestion_model.finish_init(schema.core_schema)
    resolved: dict[EdgeId, list[Any]] = {}
    resolved_count = unresolved_count = 0
    writer = DBWriter(schema=schema, ingestion_model=ingestion_model)
    try:
        await writer.write(
            gc=to_write,
            conn_conf=conn_conf,
            resource_name=None,
            stamp=stamp,
            guard=guard,
        )
        if pending:
            resolved, resolved_count, unresolved_count = await asyncio.to_thread(
                _resolve_pending, schema, conn_conf, pending, ambiguity, guard
            )
            if resolved:
                await writer.write(
                    gc=GraphContainer(edges=dict(resolved)),
                    conn_conf=conn_conf,
                    resource_name=None,
                    stamp=stamp,
                    guard=guard,
                )
    finally:
        writer.close()

    edges: dict[str, int] = {}
    for source in (to_write.edges, resolved):
        for edge_id, triples in source.items():
            label = edge_label(edge_id)
            edges[label] = edges.get(label, 0) + len(triples)
    return ContainerWriteReport(
        target=str(conn_conf.connection_type),
        schema_defined=defined,
        vertices={name: len(docs) for name, docs in to_write.vertices.items()},
        edges=edges,
        lookups_resolved=resolved_count,
        lookups_unresolved=unresolved_count,
    )