graflo.db.graflo_backend.connection

GraFlo file backend connection implementation.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

GraFloBackendConnection

Bases: Connection

Read/write graph data through a chunked on-disk GraFlo backend.

Writes append; reads fold the appended records by identity (:mod:~graflo.architecture.backend.fold), so an upsert behaves as on a database, merge: union fields included. Every read therefore scans the whole collection it reads, whatever its limit: this backend holds dumps and serves tests; it is not a database to serve a graph from.

Source code in graflo/db/graflo_backend/connection.py

class GraFloBackendConnection(Connection):
    """Read/write graph data through a chunked on-disk GraFlo backend.

    Writes append; reads fold the appended records by identity
    (:mod:`~graflo.architecture.backend.fold`), so an upsert behaves as on a
    database, ``merge: union`` fields included. Every read therefore scans the
    whole collection it reads, whatever its ``limit``: this backend holds dumps
    and serves tests; it is not a database to serve a graph from.
    """

    flavor = DBType.GRAFLO_BACKEND
    supports_graph_export = True
    supports_schema_introspection = True
    #: Introspection reads the ``schema.yaml`` catalogue the writer keeps.
    schema_introspection_is_sampled = False
    #: Guards are evaluated in process, on every document and edge row read.
    supports_read_guard = True
    supports_union_merge = True
    supports_membership_read = True
    supports_bulk_load = True

    def __init__(self, config: GraFloBackendConfig) -> None:
        super().__init__()
        self.config = config
        self._edge_index_cache: dict[str, list[dict[str, Any]]] | None = None
        self._writer = GraFloBackendWriter(
            config.output_dir,
            chunk_size=config.chunk_size,
            resume=True,
        )
        self._reader = GraFloBackendReader(config.output_dir)
        self._schema_written = False

    def create_database(self, name: str) -> None:
        self.config.output_dir.mkdir(parents=True, exist_ok=True)

    def delete_database(self, name: str) -> None:
        if self.config.output_dir.exists():
            shutil.rmtree(self.config.output_dir)

    def execute(self, query: str | Any, **kwargs: Any) -> Any:
        raise NotImplementedError("GraFlo file backend does not support ad-hoc queries")

    def close(self) -> None:
        if (
            self._schema_written
            or self._writer._vertex_writers
            or self._writer._edge_writers
        ):
            self._writer.flush_index()

    def define_schema(self, schema: Schema) -> None:
        self._writer.write_schema(schema)
        self._schema_written = True

    def delete_graph_structure(
        self,
        vertex_types: tuple[str, ...] | list[str] = (),
        graph_names: tuple[str, ...] | list[str] = (),
        delete_all: bool = False,
    ) -> None:
        if delete_all:
            self._writer.reset_data()

    def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
        if self.config.output_dir.exists():
            return
        if not create:
            raise NamespaceNotFoundError(
                f"GraFlo backend output directory {self.config.output_dir} "
                "does not exist. Create it manually or call with create_namespace=True."
            )
        self.config.output_dir.mkdir(parents=True, exist_ok=True)

    def apply_target_schema(
        self,
        schema: Schema,
        *,
        recreate: bool,
        create_namespace: bool = True,
    ) -> None:
        self.report_edge_direction_support(schema)
        layout = GraFloLayout(self.config.output_dir)
        if layout.schema_path.exists() and not recreate:
            raise SchemaExistsError(
                f"GraFlo backend already exists at {self.config.output_dir}"
            )
        if recreate and self.config.output_dir.exists():
            self._writer.reset_data()
        if create_namespace:
            self.config.output_dir.mkdir(parents=True, exist_ok=True)
        self._writer.write_schema(schema)
        self._schema_written = True

    def init_db(
        self,
        schema: Schema,
        recreate_schema: bool = False,
        *,
        create_namespace: bool = True,
    ) -> None:
        """Convenience wrapper: ensure output dir then write schema."""
        self.ensure_target_namespace(schema, create=create_namespace)
        self.apply_target_schema(
            schema, recreate=recreate_schema, create_namespace=create_namespace
        )

    def clear_data(self, schema: Schema) -> None:
        self._writer.reset_data()

    def upsert_docs_batch(
        self,
        docs: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        **kwargs: Any,
    ) -> None:
        # ``union_fields`` needs no handling here: reads fold every record by
        # the schema's identity and merge modes, the same ones it was built from.
        if kwargs.get("dry"):
            return
        self._writer.write_vertex_batch(class_name, docs)

    def insert_edges_batch(
        self,
        docs_edges: list[list[dict[str, Any]]] | list[Any] | None,
        source_class: str,
        target_class: str,
        relation_name: str,
        match_keys_source: tuple[str, ...],
        match_keys_target: tuple[str, ...],
        filter_uniques: bool = True,
        head: int | None = None,
        **kwargs: Any,
    ) -> None:
        if kwargs.get("dry") or not docs_edges:
            return
        edge_key = (source_class, target_class, relation_name or None)
        self._writer.write_edge_batch(edge_key, list(docs_edges))

    def insert_return_batch(
        self, docs: list[dict[str, Any]], class_name: str
    ) -> list[dict[str, Any]]:
        self.upsert_docs_batch(docs, class_name, match_keys=[])
        return docs

    # -- bulk load: appends, folded on read like any write ---------------------

    def bulk_begin(
        self,
        schema: Schema,
        cfg: BulkConfig,
        *,
        recreate: bool = False,
        resume: bool = False,
    ) -> BulkPlan:
        """Write the schema through the first-write guard; one writer at a time."""
        self.ensure_target_namespace(schema, create=True)
        if not resume or not GraFloLayout(self.config.output_dir).schema_path.exists():
            self.apply_target_schema(schema, recreate=recreate)
        return BulkPlan(max_parallel=1)

    def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
        # Published before returning, so a chunk recorded as written is on disk;
        # a replayed chunk appends records the read-side fold merges.
        self._writer.write_vertex_batch(chunk.class_name, chunk.rows)
        self._writer.flush_index()

    def bulk_prepare_edges(self, schema: Schema) -> None:
        """Nothing to prepare: edges name their endpoints by key."""

    def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
        edge_key = (chunk.source_class, chunk.target_class, chunk.relation_name or None)
        self._writer.write_edge_batch(edge_key, [list(row) for row in chunk.rows])
        self._writer.flush_index()

    def bulk_finish(self, schema: Schema) -> BulkCounts:
        self._sync_for_read()
        index = self._reader.read_index()
        edges: dict[StoredEdgeId, int] = {}
        for name in index.edges:
            source, target, relation = GraFloLayout.index_name_to_edge_key(name)
            edges[(source, relation, target)] = len(
                self._reader.edge_records((source, target, relation))
            )
        return BulkCounts(
            vertices={
                name: len(self._reader.vertex_records(name)) for name in index.vertices
            },
            edges=edges,
        )

    def _sync_for_read(self) -> None:
        """Persist buffered writes, and see what any writer has published since.

        Chunk writers buffer records in memory and the index is only written on
        flush, so a read issued mid-ingest would otherwise miss data — or find
        no ``INDEX.json`` at all. Every other backend is read-your-writes;
        flushing here gives the file backend the same contract. Another
        connection may write the same directory, so the index is re-read and
        the edge index dropped when it changed.
        """
        try:
            self._writer.flush_index()
        except ValueError:
            # No schema written yet, so nothing has been ingested to observe.
            pass
        if self._reader.reload_index():
            self._edge_index_cache = None

    def fetch_docs(
        self,
        class_name: str,
        filters: list[Any] | dict[str, Any] | FilterExpression | None = None,
        limit: int | None = None,
        return_keys: list[str] | None = None,
        unset_keys: list[str] | None = None,
        guard: FilterExpression | None = None,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Documents of *class_name*, one per entity, that match *filters*.

        Args:
            guard: A trusted predicate every document must also satisfy.
        """
        self._sync_for_read()
        keep = _matcher(filters, guard)

        def _project(doc: dict[str, Any]) -> dict[str, Any]:
            if return_keys:
                doc = {k: doc.get(k) for k in return_keys}
            if unset_keys:
                doc = {k: v for k, v in doc.items() if k not in unset_keys}
            return doc

        # `limit` bounds the *result* size, so filtering happens before slicing.
        docs = [
            _project(doc)
            for doc in self._reader.vertex_records(class_name)
            if keep(doc)
        ]
        return docs if limit is None else docs[:limit]

    def fetch_edges(
        self,
        from_type: str,
        from_id: str,
        edge_type: str | None = None,
        to_type: str | None = None,
        to_id: str | None = None,
        filters: list[Any] | dict[str, Any] | None = None,
        limit: int | None = None,
        return_keys: list[str] | None = None,
        unset_keys: list[str] | None = None,
        direction: EdgeDirection = EdgeDirection.OUT,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """Edges incident to one vertex, served from a lazily built index.

        Direction is this backend's storage partition key, so the reverse
        orientation has to be materialized to be answerable at all. That is a
        statement about *storage*; an in-process read index is not storage, so
        building one here does not contradict the ``MATERIALIZATION_REQUIRED``
        tier in the capability matrix — it is exactly what that tier prescribes.
        """
        self._sync_for_read()
        index = self._edge_index()
        entries = index.get(edge_type) if edge_type is not None else None
        if entries is None:
            # No edge_type: search every indexed edge type.
            entries = [row for rows in index.values() for row in rows]

        # Filtered before `limit`, which bounds the result.
        keep = _matcher(filters)
        matched: list[dict[str, Any]] = []
        for row in entries:
            source_id = row.get("_from_key")
            target_id = row.get("_to_key")
            if direction is EdgeDirection.OUT:
                anchored = source_id == from_id
                far = target_id
            elif direction is EdgeDirection.IN:
                anchored = target_id == from_id
                far = source_id
            else:
                if source_id == from_id:
                    anchored, far = True, target_id
                elif target_id == from_id:
                    anchored, far = True, source_id
                else:
                    anchored, far = False, None
            if not anchored:
                continue
            if to_id is not None and far != to_id:
                continue
            if not keep(row):
                continue
            matched.append(row)
            if limit is not None and len(matched) >= limit:
                break

        if return_keys or unset_keys:
            keep = set(return_keys) if return_keys else None
            drop = set(unset_keys) if unset_keys else set()
            matched = [
                {
                    k: v
                    for k, v in row.items()
                    if (keep is None or k in keep) and k not in drop
                }
                for row in matched
            ]
        return matched

    def _edge_index(self) -> dict[str, list[dict[str, Any]]]:
        """Edge query name -> flat edge rows, built once per connection.

        Rows carry ``_from_key`` / ``_to_key`` so both orientations are
        answerable from one pass over the chunked files.
        """
        if self._edge_index_cache is not None:
            return self._edge_index_cache

        schema = self._reader.read_schema()
        db_aware = schema.resolve_db_aware(self.flavor)
        keys = PhysicalKeys(stored_schema(schema).db_profile)
        index: dict[str, list[dict[str, Any]]] = {}
        total = 0
        for edge in schema.core_schema.edge_config.edges:
            # Keyed by the name a walk asks for the edge by.
            name = edge_query_name(db_aware, edge, self.flavor)
            if name is None:
                continue
            rows = index.setdefault(name, [])
            # Stored records carry stored property names.
            source_identity = keys.vertex_fields(
                edge.source, db_aware.vertex_config.identity_fields(edge.source)
            )
            target_identity = keys.vertex_fields(
                edge.target, db_aware.vertex_config.identity_fields(edge.target)
            )
            for record in self._reader.edge_records(
                edge_chunk_key(schema, edge.edge_id)
            ):
                if not isinstance(record, list) or len(record) < 2:
                    continue
                source_doc, target_doc = record[0], record[1]
                weight = record[2] if len(record) > 2 else {}
                rows.append(
                    {
                        **(weight if isinstance(weight, dict) else {}),
                        "_from_key": _first_value(source_doc, source_identity),
                        "_to_key": _first_value(target_doc, target_identity),
                    }
                )
                total += 1
                if total >= _EDGE_INDEX_MAX_ROWS:
                    logger.warning(
                        "GraFlo file backend edge index hit its %s-row bound; "
                        "traversal results may be incomplete",
                        _EDGE_INDEX_MAX_ROWS,
                    )
                    self._edge_index_cache = index
                    return index
        self._edge_index_cache = index
        return index

    def fetch_present_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        flatten: bool = False,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
        if not match_keys:
            return [] if flatten else {}
        existing = {
            tuple(doc.get(key) for key in match_keys): doc
            for doc in self.fetch_docs(class_name)
        }
        if flatten:
            present: list[dict[str, Any]] = []
            for doc in batch:
                key = tuple(doc.get(field) for field in match_keys)
                match = existing.get(key)
                if match is not None:
                    present.append(match)
            return present
        result: dict[int, list[dict[str, Any]]] = {}
        for index, doc in enumerate(batch):
            key = tuple(doc.get(field) for field in match_keys)
            match = existing.get(key)
            if match is not None:
                result[index] = [match]
        return result

    def aggregate(
        self,
        class_name: str,
        aggregation_function: AggregationType,
        discriminant: str | None = None,
        aggregated_field: str | None = None,
        filters: FilterExpression | list[Any] | dict[str, Any] | None = None,
    ) -> int | float | list[dict[str, Any]] | dict[str, int | float] | None:
        """Aggregate over the documents of a vertex type, read in full.

        Returns what the Cypher backends return: an ungrouped COUNT an int, a
        COUNT grouped by *discriminant* ``{value: count}``, MAX / MIN / AVERAGE
        a scalar or ``None``, SORTED_UNIQUE a list. Documents without
        *aggregated_field* are left out.

        Raises:
            ValueError: If *aggregated_field* is missing where it is required.
        """
        docs = self.fetch_docs(class_name, filters=cast(Any, filters))
        if aggregation_function == AggregationType.COUNT:
            if discriminant:
                return cast(
                    dict[str, int | float],
                    dict(Counter(doc.get(discriminant) for doc in docs)),
                )
            return len(docs)
        if not aggregated_field:
            raise ValueError(f"aggregated_field is required for {aggregation_function}")
        values = [
            doc[aggregated_field]
            for doc in docs
            if doc.get(aggregated_field) is not None
        ]
        if aggregation_function == AggregationType.SORTED_UNIQUE:
            return sorted(set(values))
        if not values:
            return None
        if aggregation_function == AggregationType.MAX:
            return max(values)
        if aggregation_function == AggregationType.MIN:
            return min(values)
        if aggregation_function == AggregationType.AVERAGE:
            return sum(values) / len(values)
        raise ValueError(f"Unsupported aggregation type: {aggregation_function}")

    def keep_absent_documents(
        self,
        batch: list[dict[str, Any]],
        class_name: str,
        match_keys: list[str] | tuple[str, ...],
        keep_keys: list[str] | tuple[str, ...] | None = None,
        filters: list[Any] | dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        if not match_keys:
            return batch
        existing = {
            tuple(doc.get(key) for key in match_keys)
            for doc in self.fetch_docs(class_name)
        }
        return [
            doc
            for doc in batch
            if tuple(doc.get(field) for field in match_keys) not in existing
        ]

    def define_vertex_indexes(
        self, vertex_config: VertexConfig, schema: Schema | None = None
    ) -> None:
        return None

    def define_edge_indexes(
        self, edges: list[Edge], schema: Schema | None = None
    ) -> None:
        return None

    def introspect_graph_schema(
        self,
        schema_name: str | None = None,
        *,
        sample_limit: int = 100,
    ) -> Schema:
        return self._reader.read_schema()

    def fetch_all_docs(
        self,
        class_name: str,
        *,
        limit: int | None = None,
    ) -> list[dict[str, Any]]:
        return self.fetch_docs(class_name, limit=limit)

    def fetch_all_edges(
        self,
        source_class: str,
        target_class: str,
        relation_name: str | None,
        *,
        match_keys_source: tuple[str, ...] | None = None,
        match_keys_target: tuple[str, ...] | None = None,
        limit: int | None = None,
        collection_name: str | None = None,
    ) -> list[list[dict[str, Any]]]:
        self._sync_for_read()
        docs = self._reader.edge_records((source_class, target_class, relation_name))
        return docs if limit is None else docs[:limit]

    def _graph_neighbors(
        self,
        vertex_type: str,
        key: str | dict[str, Any],
        *,
        hops: int = 1,
        direction: EdgeDirection = EdgeDirection.OUT,
        edge_types: Sequence[str] | None = None,
        filters: Any | None = None,
        limit: int | None = None,
        schema: Schema | None = None,
        guard: FilterExpression | None = None,
    ) -> GraphContainer:
        """The default breadth-first walk, with *guard* applied to every element."""
        return bfs_neighbors(
            self,
            anchor_type=vertex_type,
            anchor_key=key,
            hops=hops,
            direction=direction,
            edge_types=edge_types,
            filters=filters,
            limit=limit,
            schema=schema,
            guard=guard,
        )

    def export_graph_container(
        self, schema: Schema, *, limit: int | None = None
    ) -> GraphContainer:
        """The graph as :meth:`GraFloBackendReader.load_graph_container` reads it.

        Keyed by the logical names of ``schema.yaml``, which is also the schema
        :meth:`introspect_graph_schema` returns.
        """
        self._sync_for_read()
        return self._reader.load_graph_container(limit=limit)

Attributes

config = config instance-attribute

flavor = DBType.GRAFLO_BACKEND class-attribute instance-attribute

schema_introspection_is_sampled = False class-attribute instance-attribute

supports_bulk_load = True class-attribute instance-attribute

supports_graph_export = True class-attribute instance-attribute

supports_membership_read = True class-attribute instance-attribute

supports_read_guard = True class-attribute instance-attribute

supports_schema_introspection = True class-attribute instance-attribute

supports_union_merge = True class-attribute instance-attribute

Methods:

__init__(config)

Source code in graflo/db/graflo_backend/connection.py

def __init__(self, config: GraFloBackendConfig) -> None:
    super().__init__()
    self.config = config
    self._edge_index_cache: dict[str, list[dict[str, Any]]] | None = None
    self._writer = GraFloBackendWriter(
        config.output_dir,
        chunk_size=config.chunk_size,
        resume=True,
    )
    self._reader = GraFloBackendReader(config.output_dir)
    self._schema_written = False

aggregate(class_name, aggregation_function, discriminant=None, aggregated_field=None, filters=None)

Aggregate over the documents of a vertex type, read in full.

Returns what the Cypher backends return: an ungrouped COUNT an int, a COUNT grouped by discriminant {value: count}, MAX / MIN / AVERAGE a scalar or None, SORTED_UNIQUE a list. Documents without aggregated_field are left out.

Raises:

Type	Description
ValueError	If aggregated_field is missing where it is required.

Source code in graflo/db/graflo_backend/connection.py

def aggregate(
    self,
    class_name: str,
    aggregation_function: AggregationType,
    discriminant: str | None = None,
    aggregated_field: str | None = None,
    filters: FilterExpression | list[Any] | dict[str, Any] | None = None,
) -> int | float | list[dict[str, Any]] | dict[str, int | float] | None:
    """Aggregate over the documents of a vertex type, read in full.

    Returns what the Cypher backends return: an ungrouped COUNT an int, a
    COUNT grouped by *discriminant* ``{value: count}``, MAX / MIN / AVERAGE
    a scalar or ``None``, SORTED_UNIQUE a list. Documents without
    *aggregated_field* are left out.

    Raises:
        ValueError: If *aggregated_field* is missing where it is required.
    """
    docs = self.fetch_docs(class_name, filters=cast(Any, filters))
    if aggregation_function == AggregationType.COUNT:
        if discriminant:
            return cast(
                dict[str, int | float],
                dict(Counter(doc.get(discriminant) for doc in docs)),
            )
        return len(docs)
    if not aggregated_field:
        raise ValueError(f"aggregated_field is required for {aggregation_function}")
    values = [
        doc[aggregated_field]
        for doc in docs
        if doc.get(aggregated_field) is not None
    ]
    if aggregation_function == AggregationType.SORTED_UNIQUE:
        return sorted(set(values))
    if not values:
        return None
    if aggregation_function == AggregationType.MAX:
        return max(values)
    if aggregation_function == AggregationType.MIN:
        return min(values)
    if aggregation_function == AggregationType.AVERAGE:
        return sum(values) / len(values)
    raise ValueError(f"Unsupported aggregation type: {aggregation_function}")

apply_target_schema(schema, *, recreate, create_namespace=True)

Source code in graflo/db/graflo_backend/connection.py

def apply_target_schema(
    self,
    schema: Schema,
    *,
    recreate: bool,
    create_namespace: bool = True,
) -> None:
    self.report_edge_direction_support(schema)
    layout = GraFloLayout(self.config.output_dir)
    if layout.schema_path.exists() and not recreate:
        raise SchemaExistsError(
            f"GraFlo backend already exists at {self.config.output_dir}"
        )
    if recreate and self.config.output_dir.exists():
        self._writer.reset_data()
    if create_namespace:
        self.config.output_dir.mkdir(parents=True, exist_ok=True)
    self._writer.write_schema(schema)
    self._schema_written = True

bulk_begin(schema, cfg, *, recreate=False, resume=False)

Write the schema through the first-write guard; one writer at a time.

Source code in graflo/db/graflo_backend/connection.py

def bulk_begin(
    self,
    schema: Schema,
    cfg: BulkConfig,
    *,
    recreate: bool = False,
    resume: bool = False,
) -> BulkPlan:
    """Write the schema through the first-write guard; one writer at a time."""
    self.ensure_target_namespace(schema, create=True)
    if not resume or not GraFloLayout(self.config.output_dir).schema_path.exists():
        self.apply_target_schema(schema, recreate=recreate)
    return BulkPlan(max_parallel=1)

bulk_finish(schema)

Source code in graflo/db/graflo_backend/connection.py

def bulk_finish(self, schema: Schema) -> BulkCounts:
    self._sync_for_read()
    index = self._reader.read_index()
    edges: dict[StoredEdgeId, int] = {}
    for name in index.edges:
        source, target, relation = GraFloLayout.index_name_to_edge_key(name)
        edges[(source, relation, target)] = len(
            self._reader.edge_records((source, target, relation))
        )
    return BulkCounts(
        vertices={
            name: len(self._reader.vertex_records(name)) for name in index.vertices
        },
        edges=edges,
    )

bulk_prepare_edges(schema)

Nothing to prepare: edges name their endpoints by key.

Source code in graflo/db/graflo_backend/connection.py

def bulk_prepare_edges(self, schema: Schema) -> None:
    """Nothing to prepare: edges name their endpoints by key."""

bulk_write_edges(schema, chunk)

Source code in graflo/db/graflo_backend/connection.py

def bulk_write_edges(self, schema: Schema, chunk: BulkEdgeChunk) -> None:
    edge_key = (chunk.source_class, chunk.target_class, chunk.relation_name or None)
    self._writer.write_edge_batch(edge_key, [list(row) for row in chunk.rows])
    self._writer.flush_index()

bulk_write_vertices(schema, chunk)

Source code in graflo/db/graflo_backend/connection.py

def bulk_write_vertices(self, schema: Schema, chunk: BulkVertexChunk) -> None:
    # Published before returning, so a chunk recorded as written is on disk;
    # a replayed chunk appends records the read-side fold merges.
    self._writer.write_vertex_batch(chunk.class_name, chunk.rows)
    self._writer.flush_index()

clear_data(schema)

Source code in graflo/db/graflo_backend/connection.py

def clear_data(self, schema: Schema) -> None:
    self._writer.reset_data()

close()

Source code in graflo/db/graflo_backend/connection.py

def close(self) -> None:
    if (
        self._schema_written
        or self._writer._vertex_writers
        or self._writer._edge_writers
    ):
        self._writer.flush_index()

create_database(name)

Source code in graflo/db/graflo_backend/connection.py

def create_database(self, name: str) -> None:
    self.config.output_dir.mkdir(parents=True, exist_ok=True)

define_edge_indexes(edges, schema=None)

Source code in graflo/db/graflo_backend/connection.py

def define_edge_indexes(
    self, edges: list[Edge], schema: Schema | None = None
) -> None:
    return None

define_schema(schema)

Source code in graflo/db/graflo_backend/connection.py

def define_schema(self, schema: Schema) -> None:
    self._writer.write_schema(schema)
    self._schema_written = True

define_vertex_indexes(vertex_config, schema=None)

Source code in graflo/db/graflo_backend/connection.py

def define_vertex_indexes(
    self, vertex_config: VertexConfig, schema: Schema | None = None
) -> None:
    return None

delete_database(name)

Source code in graflo/db/graflo_backend/connection.py

def delete_database(self, name: str) -> None:
    if self.config.output_dir.exists():
        shutil.rmtree(self.config.output_dir)

delete_graph_structure(vertex_types=(), graph_names=(), delete_all=False)

Source code in graflo/db/graflo_backend/connection.py

def delete_graph_structure(
    self,
    vertex_types: tuple[str, ...] | list[str] = (),
    graph_names: tuple[str, ...] | list[str] = (),
    delete_all: bool = False,
) -> None:
    if delete_all:
        self._writer.reset_data()

ensure_target_namespace(schema, *, create)

Source code in graflo/db/graflo_backend/connection.py

def ensure_target_namespace(self, schema: Schema, *, create: bool) -> None:
    if self.config.output_dir.exists():
        return
    if not create:
        raise NamespaceNotFoundError(
            f"GraFlo backend output directory {self.config.output_dir} "
            "does not exist. Create it manually or call with create_namespace=True."
        )
    self.config.output_dir.mkdir(parents=True, exist_ok=True)

execute(query, **kwargs)

Source code in graflo/db/graflo_backend/connection.py

def execute(self, query: str | Any, **kwargs: Any) -> Any:
    raise NotImplementedError("GraFlo file backend does not support ad-hoc queries")

export_graph_container(schema, *, limit=None)

The graph as :meth:GraFloBackendReader.load_graph_container reads it.

Keyed by the logical names of schema.yaml, which is also the schema :meth:introspect_graph_schema returns.

Source code in graflo/db/graflo_backend/connection.py

def export_graph_container(
    self, schema: Schema, *, limit: int | None = None
) -> GraphContainer:
    """The graph as :meth:`GraFloBackendReader.load_graph_container` reads it.

    Keyed by the logical names of ``schema.yaml``, which is also the schema
    :meth:`introspect_graph_schema` returns.
    """
    self._sync_for_read()
    return self._reader.load_graph_container(limit=limit)

fetch_all_docs(class_name, *, limit=None)

Source code in graflo/db/graflo_backend/connection.py

def fetch_all_docs(
    self,
    class_name: str,
    *,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    return self.fetch_docs(class_name, limit=limit)

fetch_all_edges(source_class, target_class, relation_name, *, match_keys_source=None, match_keys_target=None, limit=None, collection_name=None)

Source code in graflo/db/graflo_backend/connection.py

def fetch_all_edges(
    self,
    source_class: str,
    target_class: str,
    relation_name: str | None,
    *,
    match_keys_source: tuple[str, ...] | None = None,
    match_keys_target: tuple[str, ...] | None = None,
    limit: int | None = None,
    collection_name: str | None = None,
) -> list[list[dict[str, Any]]]:
    self._sync_for_read()
    docs = self._reader.edge_records((source_class, target_class, relation_name))
    return docs if limit is None else docs[:limit]

fetch_docs(class_name, filters=None, limit=None, return_keys=None, unset_keys=None, guard=None, **kwargs)

Documents of class_name, one per entity, that match filters.

Parameters:

Name	Type	Description	Default
guard	FilterExpression | None	A trusted predicate every document must also satisfy.	None

Source code in graflo/db/graflo_backend/connection.py

def fetch_docs(
    self,
    class_name: str,
    filters: list[Any] | dict[str, Any] | FilterExpression | None = None,
    limit: int | None = None,
    return_keys: list[str] | None = None,
    unset_keys: list[str] | None = None,
    guard: FilterExpression | None = None,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """Documents of *class_name*, one per entity, that match *filters*.

    Args:
        guard: A trusted predicate every document must also satisfy.
    """
    self._sync_for_read()
    keep = _matcher(filters, guard)

    def _project(doc: dict[str, Any]) -> dict[str, Any]:
        if return_keys:
            doc = {k: doc.get(k) for k in return_keys}
        if unset_keys:
            doc = {k: v for k, v in doc.items() if k not in unset_keys}
        return doc

    # `limit` bounds the *result* size, so filtering happens before slicing.
    docs = [
        _project(doc)
        for doc in self._reader.vertex_records(class_name)
        if keep(doc)
    ]
    return docs if limit is None else docs[:limit]

fetch_edges(from_type, from_id, edge_type=None, to_type=None, to_id=None, filters=None, limit=None, return_keys=None, unset_keys=None, direction=EdgeDirection.OUT, **kwargs)

Edges incident to one vertex, served from a lazily built index.

Direction is this backend's storage partition key, so the reverse orientation has to be materialized to be answerable at all. That is a statement about storage; an in-process read index is not storage, so building one here does not contradict the MATERIALIZATION_REQUIRED tier in the capability matrix — it is exactly what that tier prescribes.

Source code in graflo/db/graflo_backend/connection.py

def fetch_edges(
    self,
    from_type: str,
    from_id: str,
    edge_type: str | None = None,
    to_type: str | None = None,
    to_id: str | None = None,
    filters: list[Any] | dict[str, Any] | None = None,
    limit: int | None = None,
    return_keys: list[str] | None = None,
    unset_keys: list[str] | None = None,
    direction: EdgeDirection = EdgeDirection.OUT,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """Edges incident to one vertex, served from a lazily built index.

    Direction is this backend's storage partition key, so the reverse
    orientation has to be materialized to be answerable at all. That is a
    statement about *storage*; an in-process read index is not storage, so
    building one here does not contradict the ``MATERIALIZATION_REQUIRED``
    tier in the capability matrix — it is exactly what that tier prescribes.
    """
    self._sync_for_read()
    index = self._edge_index()
    entries = index.get(edge_type) if edge_type is not None else None
    if entries is None:
        # No edge_type: search every indexed edge type.
        entries = [row for rows in index.values() for row in rows]

    # Filtered before `limit`, which bounds the result.
    keep = _matcher(filters)
    matched: list[dict[str, Any]] = []
    for row in entries:
        source_id = row.get("_from_key")
        target_id = row.get("_to_key")
        if direction is EdgeDirection.OUT:
            anchored = source_id == from_id
            far = target_id
        elif direction is EdgeDirection.IN:
            anchored = target_id == from_id
            far = source_id
        else:
            if source_id == from_id:
                anchored, far = True, target_id
            elif target_id == from_id:
                anchored, far = True, source_id
            else:
                anchored, far = False, None
        if not anchored:
            continue
        if to_id is not None and far != to_id:
            continue
        if not keep(row):
            continue
        matched.append(row)
        if limit is not None and len(matched) >= limit:
            break

    if return_keys or unset_keys:
        keep = set(return_keys) if return_keys else None
        drop = set(unset_keys) if unset_keys else set()
        matched = [
            {
                k: v
                for k, v in row.items()
                if (keep is None or k in keep) and k not in drop
            }
            for row in matched
        ]
    return matched

fetch_present_documents(batch, class_name, match_keys, keep_keys=None, flatten=False, filters=None)

Source code in graflo/db/graflo_backend/connection.py

def fetch_present_documents(
    self,
    batch: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    keep_keys: list[str] | tuple[str, ...] | None = None,
    flatten: bool = False,
    filters: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]] | dict[int, list[dict[str, Any]]]:
    if not match_keys:
        return [] if flatten else {}
    existing = {
        tuple(doc.get(key) for key in match_keys): doc
        for doc in self.fetch_docs(class_name)
    }
    if flatten:
        present: list[dict[str, Any]] = []
        for doc in batch:
            key = tuple(doc.get(field) for field in match_keys)
            match = existing.get(key)
            if match is not None:
                present.append(match)
        return present
    result: dict[int, list[dict[str, Any]]] = {}
    for index, doc in enumerate(batch):
        key = tuple(doc.get(field) for field in match_keys)
        match = existing.get(key)
        if match is not None:
            result[index] = [match]
    return result

init_db(schema, recreate_schema=False, *, create_namespace=True)

Convenience wrapper: ensure output dir then write schema.

Source code in graflo/db/graflo_backend/connection.py

def init_db(
    self,
    schema: Schema,
    recreate_schema: bool = False,
    *,
    create_namespace: bool = True,
) -> None:
    """Convenience wrapper: ensure output dir then write schema."""
    self.ensure_target_namespace(schema, create=create_namespace)
    self.apply_target_schema(
        schema, recreate=recreate_schema, create_namespace=create_namespace
    )

insert_edges_batch(docs_edges, source_class, target_class, relation_name, match_keys_source, match_keys_target, filter_uniques=True, head=None, **kwargs)

Source code in graflo/db/graflo_backend/connection.py

def insert_edges_batch(
    self,
    docs_edges: list[list[dict[str, Any]]] | list[Any] | None,
    source_class: str,
    target_class: str,
    relation_name: str,
    match_keys_source: tuple[str, ...],
    match_keys_target: tuple[str, ...],
    filter_uniques: bool = True,
    head: int | None = None,
    **kwargs: Any,
) -> None:
    if kwargs.get("dry") or not docs_edges:
        return
    edge_key = (source_class, target_class, relation_name or None)
    self._writer.write_edge_batch(edge_key, list(docs_edges))

insert_return_batch(docs, class_name)

Source code in graflo/db/graflo_backend/connection.py

def insert_return_batch(
    self, docs: list[dict[str, Any]], class_name: str
) -> list[dict[str, Any]]:
    self.upsert_docs_batch(docs, class_name, match_keys=[])
    return docs

introspect_graph_schema(schema_name=None, *, sample_limit=100)

Source code in graflo/db/graflo_backend/connection.py

def introspect_graph_schema(
    self,
    schema_name: str | None = None,
    *,
    sample_limit: int = 100,
) -> Schema:
    return self._reader.read_schema()

keep_absent_documents(batch, class_name, match_keys, keep_keys=None, filters=None)

Source code in graflo/db/graflo_backend/connection.py

def keep_absent_documents(
    self,
    batch: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    keep_keys: list[str] | tuple[str, ...] | None = None,
    filters: list[Any] | dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    if not match_keys:
        return batch
    existing = {
        tuple(doc.get(key) for key in match_keys)
        for doc in self.fetch_docs(class_name)
    }
    return [
        doc
        for doc in batch
        if tuple(doc.get(field) for field in match_keys) not in existing
    ]

upsert_docs_batch(docs, class_name, match_keys, **kwargs)

Source code in graflo/db/graflo_backend/connection.py

def upsert_docs_batch(
    self,
    docs: list[dict[str, Any]],
    class_name: str,
    match_keys: list[str] | tuple[str, ...],
    **kwargs: Any,
) -> None:
    # ``union_fields`` needs no handling here: reads fold every record by
    # the schema's identity and merge modes, the same ones it was built from.
    if kwargs.get("dry"):
        return
    self._writer.write_vertex_batch(class_name, docs)

Functions: