Skip to content

graflo.migrate.store

Migration history persistence.

Classes

FileMigrationStore

File-backed migration history store.

Source code in graflo/migrate/store.py 
11
12
13
14
15
16
17
18
19
20
21
22
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
class FileMigrationStore:
    """File-backed migration history store."""

    def __init__(self, path: str | Path = ".graflo/migrations.json"):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        if not self.path.exists():
            self._write({"records": []})

    def history(self) -> list[MigrationRecord]:
        payload = self._read()
        records = payload.get("records", [])
        return [MigrationRecord.model_validate(item) for item in records]

    def has_revision(self, revision: str, backend: str) -> bool:
        return any(
            record.revision == revision and record.backend == backend
            for record in self.history()
        )

    def get_revision(self, revision: str, backend: str) -> MigrationRecord | None:
        for record in self.history():
            if record.revision == revision and record.backend == backend:
                return record
        return None

    def has_schema_hash(self, schema_hash: str, backend: str) -> bool:
        return any(
            record.schema_hash == schema_hash and record.backend == backend
            for record in self.history()
        )

    def add_record(self, record: MigrationRecord) -> None:
        payload = self._read()
        records = payload.get("records", [])
        records.append(record.model_dump())
        payload["records"] = records
        self._write(payload)

    def latest(self, backend: str | None = None) -> MigrationRecord | None:
        records = self.history()
        if backend is not None:
            records = [record for record in records if record.backend == backend]
        if not records:
            return None
        return records[-1]

    def _read(self) -> dict:
        if not self.path.exists():
            return {"records": []}
        raw = self.path.read_text(encoding="utf-8")
        if not raw.strip():
            return {"records": []}
        return json.loads(raw)

    def _write(self, payload: dict) -> None:
        self.path.write_text(
            json.dumps(payload, indent=2, sort_keys=True),
            encoding="utf-8",
        )

Attributes

path = Path(path) instance-attribute

Methods:

__init__(path='.graflo/migrations.json')
Source code in graflo/migrate/store.py 
14
15
16
17
18
def __init__(self, path: str | Path = ".graflo/migrations.json"):
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    if not self.path.exists():
        self._write({"records": []})
add_record(record)
Source code in graflo/migrate/store.py 
43
44
45
46
47
48
def add_record(self, record: MigrationRecord) -> None:
    payload = self._read()
    records = payload.get("records", [])
    records.append(record.model_dump())
    payload["records"] = records
    self._write(payload)
get_revision(revision, backend)
Source code in graflo/migrate/store.py 
31
32
33
34
35
def get_revision(self, revision: str, backend: str) -> MigrationRecord | None:
    for record in self.history():
        if record.revision == revision and record.backend == backend:
            return record
    return None
has_revision(revision, backend)
Source code in graflo/migrate/store.py 
25
26
27
28
29
def has_revision(self, revision: str, backend: str) -> bool:
    return any(
        record.revision == revision and record.backend == backend
        for record in self.history()
    )
has_schema_hash(schema_hash, backend)
Source code in graflo/migrate/store.py 
37
38
39
40
41
def has_schema_hash(self, schema_hash: str, backend: str) -> bool:
    return any(
        record.schema_hash == schema_hash and record.backend == backend
        for record in self.history()
    )
history()
Source code in graflo/migrate/store.py 
20
21
22
23
def history(self) -> list[MigrationRecord]:
    payload = self._read()
    records = payload.get("records", [])
    return [MigrationRecord.model_validate(item) for item in records]
latest(backend=None)
Source code in graflo/migrate/store.py 
50
51
52
53
54
55
56
def latest(self, backend: str | None = None) -> MigrationRecord | None:
    records = self.history()
    if backend is not None:
        records = [record for record in records if record.backend == backend]
    if not records:
        return None
    return records[-1]