Skip to content

graflo.architecture.backend.layout

Path conventions for GraFlo file backend directories.

GraFloLayout

Deterministic path builder for a GraFlo backend root directory.

Source code in graflo/architecture/backend/layout.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class GraFloLayout:
    """Deterministic path builder for a GraFlo backend root directory."""

    def __init__(self, root: Path) -> None:
        self.root = root.resolve()

    @property
    def index_path(self) -> Path:
        return self.root / INDEX_FILENAME

    @property
    def schema_path(self) -> Path:
        return self.root / SCHEMA_FILENAME

    @property
    def vertices_dir(self) -> Path:
        return self.root / VERTICES_DIR

    @property
    def edges_dir(self) -> Path:
        return self.root / EDGES_DIR

    def ensure_dirs(self) -> None:
        self.vertices_dir.mkdir(parents=True, exist_ok=True)
        self.edges_dir.mkdir(parents=True, exist_ok=True)

    def vertex_chunk_path(self, vertex_type: str, chunk_index: int) -> Path:
        stem = self._vertex_stem(vertex_type)
        filename = f"{stem}.{chunk_index:03d}{CHUNK_SUFFIX}"
        return self.vertices_dir / filename

    def edge_chunk_path(
        self, edge_key: tuple[str, str, str | None], chunk_index: int
    ) -> Path:
        stem = self._edge_stem(edge_key)
        filename = f"{stem}.{chunk_index:03d}{CHUNK_SUFFIX}"
        return self.edges_dir / filename

    def relative_vertex_chunk(self, vertex_type: str, chunk_index: int) -> str:
        return (
            self.vertex_chunk_path(vertex_type, chunk_index)
            .relative_to(self.root)
            .as_posix()
        )

    def relative_edge_chunk(
        self, edge_key: tuple[str, str, str | None], chunk_index: int
    ) -> str:
        return (
            self.edge_chunk_path(edge_key, chunk_index)
            .relative_to(self.root)
            .as_posix()
        )

    @staticmethod
    def edge_key_to_index_name(edge_key: tuple[str, str, str | None]) -> str:
        """Human-readable edge key when safe, otherwise JSON-array encoding."""
        source, target, relation = edge_key
        if (
            _SAFE_NAME_RE.fullmatch(source)
            and _SAFE_NAME_RE.fullmatch(target)
            and (relation is None or _SAFE_NAME_RE.fullmatch(relation))
        ):
            rel = relation or ""
            return f"{source}{_EDGE_DELIM}{rel}{_EDGE_DELIM}{target}"
        return serialize_edge_key(edge_key)

    @staticmethod
    def index_name_to_edge_key(name: str) -> tuple[str, str, str | None]:
        if name.startswith("["):
            return deserialize_edge_key(name)
        parts = name.split(_EDGE_DELIM)
        if len(parts) != 3:
            raise ValueError(f"Invalid edge index name: {name!r}")
        source, relation, target = parts
        return (source, target, relation or None)

    @staticmethod
    def _vertex_stem(vertex_type: str) -> str:
        if _SAFE_NAME_RE.fullmatch(vertex_type):
            return vertex_type
        return GraFloLayout._encode_stem(vertex_type)

    @staticmethod
    def _edge_stem(edge_key: tuple[str, str, str | None]) -> str:
        index_name = GraFloLayout.edge_key_to_index_name(edge_key)
        if _SAFE_NAME_RE.fullmatch(index_name.replace(_EDGE_DELIM, "_")):
            return index_name
        return GraFloLayout._encode_stem(serialize_edge_key(edge_key))

    @staticmethod
    def _encode_stem(value: str) -> str:
        encoded = base64.urlsafe_b64encode(value.encode("utf-8")).decode("ascii")
        return encoded.rstrip("=")

edge_key_to_index_name(edge_key) staticmethod

Human-readable edge key when safe, otherwise JSON-array encoding.

Source code in graflo/architecture/backend/layout.py 
77
78
79
80
81
82
83
84
85
86
87
88
@staticmethod
def edge_key_to_index_name(edge_key: tuple[str, str, str | None]) -> str:
    """Human-readable edge key when safe, otherwise JSON-array encoding."""
    source, target, relation = edge_key
    if (
        _SAFE_NAME_RE.fullmatch(source)
        and _SAFE_NAME_RE.fullmatch(target)
        and (relation is None or _SAFE_NAME_RE.fullmatch(relation))
    ):
        rel = relation or ""
        return f"{source}{_EDGE_DELIM}{rel}{_EDGE_DELIM}{target}"
    return serialize_edge_key(edge_key)