Skip to content

graflo.architecture.backend.layout

Path conventions for GraFlo file backend directories.

Attributes

CHUNK_SUFFIX = '.jsonl.gz' module-attribute

EDGES_DIR = 'edges' module-attribute

INDEX_FILENAME = 'INDEX.json' module-attribute

SCHEMA_FILENAME = 'schema.yaml' module-attribute

VERTICES_DIR = 'vertices' module-attribute

Classes

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("=")

Attributes

edges_dir property
index_path property
root = root.resolve() instance-attribute
schema_path property
vertices_dir property

Methods:

__init__(root)
Source code in graflo/architecture/backend/layout.py 
26
27
def __init__(self, root: Path) -> None:
    self.root = root.resolve()
edge_chunk_path(edge_key, chunk_index)
Source code in graflo/architecture/backend/layout.py 
54
55
56
57
58
59
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
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)
ensure_dirs()
Source code in graflo/architecture/backend/layout.py 
45
46
47
def ensure_dirs(self) -> None:
    self.vertices_dir.mkdir(parents=True, exist_ok=True)
    self.edges_dir.mkdir(parents=True, exist_ok=True)
index_name_to_edge_key(name) staticmethod
Source code in graflo/architecture/backend/layout.py 
90
91
92
93
94
95
96
97
98
@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)
relative_edge_chunk(edge_key, chunk_index)
Source code in graflo/architecture/backend/layout.py 
68
69
70
71
72
73
74
75
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()
    )
relative_vertex_chunk(vertex_type, chunk_index)
Source code in graflo/architecture/backend/layout.py 
61
62
63
64
65
66
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()
    )
vertex_chunk_path(vertex_type, chunk_index)
Source code in graflo/architecture/backend/layout.py 
49
50
51
52
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

Functions: