documentlanguagemodel Public

Watch 0 Fork 0 Star 0
Python · 2595 bytes Raw Blame History
1 """Frontmatter migration registry.
2
3 Each `dlm_version` bump registers one module under this package,
4 exporting a `migrate(raw: dict) -> dict` function that rewrites the
5 raw YAML dict from version N to version N+1. The dispatcher
6 (`dispatch.apply_pending`) chains them to walk an old document up to
7 `CURRENT_SCHEMA_VERSION`.
8
9 Enforcement contract: `test_all_versions_have_migrator_up_to_latest`
10 walks this registry and refuses any CI run where
11 `set(MIGRATORS) != set(range(1, CURRENT_SCHEMA_VERSION))`. A PR that
12 bumps the schema version without registering a matching migrator fails
13 that test, so the rule is statically enforced.
14
15 Minimum viable registration:
16
17 # src/dlm/doc/migrations/vN.py
18 def migrate(raw: dict) -> dict:
19 # rewrite `raw` from version N to version N+1
20 return {**raw, "new_field": "default"}
21
22 # src/dlm/doc/migrations/__init__.py
23 MIGRATORS[N] = vN.migrate
24 """
25
26 from __future__ import annotations
27
28 from collections.abc import Callable
29 from typing import Final
30
31 # Keep the per-version modules imported here so the shipped registry is
32 # declared in one explicit place rather than assembled via import-time
33 # side effects.
34 from dlm.doc.migrations import (
35 v1,
36 v2,
37 v3,
38 v4,
39 v5,
40 v6,
41 v7,
42 v8,
43 v9,
44 v10,
45 v11,
46 v12,
47 v13,
48 v14,
49 )
50
51 # Map of `from_version` → migrator function for the shipped schema path.
52 MIGRATORS: Final[dict[int, Callable[[dict[str, object]], dict[str, object]]]] = {
53 1: v1.migrate,
54 2: v2.migrate,
55 3: v3.migrate,
56 4: v4.migrate,
57 5: v5.migrate,
58 6: v6.migrate,
59 7: v7.migrate,
60 8: v8.migrate,
61 9: v9.migrate,
62 10: v10.migrate,
63 11: v11.migrate,
64 12: v12.migrate,
65 13: v13.migrate,
66 14: v14.migrate,
67 }
68
69
70 def register(
71 from_version: int,
72 ) -> Callable[
73 [Callable[[dict[str, object]], dict[str, object]]],
74 Callable[[dict[str, object]], dict[str, object]],
75 ]:
76 """Test helper: bind a temporary migrator to a `from_version` key.
77
78 Runtime migrators are declared explicitly in `MIGRATORS` above.
79 Tests still use this helper to swap in synthetic upgrade chains
80 without editing the shipped modules.
81 """
82
83 def decorator(
84 fn: Callable[[dict[str, object]], dict[str, object]],
85 ) -> Callable[[dict[str, object]], dict[str, object]]:
86 assert from_version not in MIGRATORS, (
87 f"duplicate migrator for from_version={from_version}: "
88 f"{MIGRATORS[from_version].__module__} vs {fn.__module__}"
89 )
90 MIGRATORS[from_version] = fn
91 return fn
92
93 return decorator