orchestrata_ultimate.session.kg_sync._common

Reference › orchestrata_ultimate › session › kg_sync › _common

Shared constants of the knowledge-graph syncer and the SyncStats dataclass. SyncStats records the outcome of one stage: the stage name, nodes created, nodes updated, relationships created, a list of error strings and the duration in milliseconds. ok is true when there are no errors, and to_dict gives the dictionary form with the duration rounded to one decimal. Every stage of the syncer returns one.

At a glance

Class diagram

Dependency diagram

API

Classes

SyncStats (dataclass)

Decorators: dataclass · Line 38

Statistics for a single sync stage.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/kg_sync/_common.py

# mypy: ignore-errors
"""Shared constants and the SyncStats dataclass for the KG syncer.

These are imported by the stage mixins and re-exported from the package
``__init__`` to preserve the original module-level public surface.
"""

from dataclasses import dataclass, field

IGNORE_DIRS = {
    "__pycache__",
    ".git",
    ".venv",
    "node_modules",
    ".tox",
    ".mypy_cache",
    ".pytest_cache",
    "dist",
    "build",
}
CODE_EXTENSIONS = {".py", ".pl"}
SYNC_EXTENSIONS = {
    ".py",
    ".pl",
    ".md",
    ".yaml",
    ".yml",
    ".json",
    ".toml",
    ".cfg",
    ".ini",
    ".txt",
}
BATCH_SIZE = 50


@dataclass
class SyncStats:
    """Statistics for a single sync stage."""

    stage: str
    nodes_created: int = 0
    nodes_updated: int = 0
    relationships_created: int = 0
    errors: list[str] = field(default_factory=list)
    duration_ms: float = 0.0

    @property
    def ok(self) -> bool:
        return len(self.errors) == 0

    def to_dict(self) -> dict:
        return {
            "stage": self.stage,
            "nodes_created": self.nodes_created,
            "nodes_updated": self.nodes_updated,
            "relationships_created": self.relationships_created,
            "errors": self.errors,
            "duration_ms": round(self.duration_ms, 1),
        }