Coverage for src/lexigram/admin/services/revisions.py: 100%
92 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Revision history service for lexigram-admin.
3Stores full field snapshots on each save and provides field-by-field diff
4comparison between any two revisions. Supports revert to any prior revision.
6The service uses a structural ``_RevisionStore`` protocol so it works with
7an in-memory store (default, zero dependencies) or a DB-backed store
8injected via the DI container.
9"""
11from __future__ import annotations
13from dataclasses import dataclass, field
14from datetime import UTC, datetime
15from typing import Any, Protocol
17# ---------------------------------------------------------------------------
18# Data models
19# ---------------------------------------------------------------------------
22@dataclass
23class Revision:
24 """A single point-in-time snapshot of a resource record.
26 Attributes:
27 revision_id: Unique identifier for this revision.
28 resource_type: Name of the resource type (e.g. ``"user"``).
29 resource_id: Identifier of the affected record.
30 data: Full field snapshot at this point in time.
31 actor_id: Who created this revision.
32 created_at: UTC timestamp.
33 comment: Optional human-readable description of the change.
34 """
36 revision_id: str
37 resource_type: str
38 resource_id: str
39 data: dict[str, Any]
40 actor_id: str
41 created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
42 comment: str = ""
45@dataclass
46class FieldDiff:
47 """The change for a single field between two revisions.
49 Attributes:
50 field_name: Name of the field that changed.
51 old_value: Value in the earlier revision (``None`` if field was absent).
52 new_value: Value in the later revision (``None`` if field was removed).
53 changed: Whether old and new values differ.
54 """
56 field_name: str
57 old_value: Any
58 new_value: Any
59 changed: bool
62@dataclass
63class RevisionDiff:
64 """Diff result between two revisions.
66 Attributes:
67 from_revision: Earlier revision identifier.
68 to_revision: Later revision identifier.
69 fields: Per-field diff entries (only fields where ``changed=True`` by default).
70 all_fields: When ``True`` all fields were included, not just changed ones.
71 """
73 from_revision: str
74 to_revision: str
75 fields: list[FieldDiff]
76 all_fields: bool = False
79# ---------------------------------------------------------------------------
80# Storage protocol
81# ---------------------------------------------------------------------------
84class _RevisionStore(Protocol):
85 """Minimal storage protocol for revision persistence."""
87 async def save(self, revision: Revision) -> None:
88 """Persist a new revision."""
89 ...
91 async def list_for_record(
92 self,
93 resource_type: str,
94 resource_id: str,
95 *,
96 limit: int = 50,
97 ) -> list[Revision]:
98 """Return revisions for a record, newest-first."""
99 ...
101 async def get(self, revision_id: str) -> Revision | None:
102 """Return a single revision by ID."""
103 ...
105 async def delete_for_record(self, resource_type: str, resource_id: str) -> int:
106 """Delete all revisions for a record. Returns count deleted."""
107 ...
110# ---------------------------------------------------------------------------
111# In-memory store (default — zero infra deps)
112# ---------------------------------------------------------------------------
115class InMemoryRevisionStore:
116 """Thread-safe in-process revision store.
118 Suitable for development and testing. Data is lost on restart.
119 For production wire a DB-backed store via the DI container.
120 """
122 def __init__(self, max_per_record: int = 100) -> None:
123 self._store: dict[str, list[Revision]] = {} # key: "type:id"
124 self._by_id: dict[str, Revision] = {}
125 self.max_per_record = max_per_record
127 def _key(self, resource_type: str, resource_id: str) -> str:
128 return f"{resource_type}:{resource_id}"
130 async def save(self, revision: Revision) -> None:
131 """Persist revision, trimming oldest if over max_per_record."""
132 k = self._key(revision.resource_type, revision.resource_id)
133 bucket = self._store.setdefault(k, [])
134 bucket.insert(0, revision)
135 self._by_id[revision.revision_id] = revision
136 if len(bucket) > self.max_per_record:
137 dropped = bucket.pop()
138 self._by_id.pop(dropped.revision_id, None)
140 async def list_for_record(
141 self,
142 resource_type: str,
143 resource_id: str,
144 *,
145 limit: int = 50,
146 ) -> list[Revision]:
147 """Return up to *limit* revisions, newest-first."""
148 k = self._key(resource_type, resource_id)
149 return self._store.get(k, [])[:limit]
151 async def get(self, revision_id: str) -> Revision | None:
152 return self._by_id.get(revision_id)
154 async def delete_for_record(self, resource_type: str, resource_id: str) -> int:
155 k = self._key(resource_type, resource_id)
156 revisions = self._store.pop(k, [])
157 for rev in revisions:
158 self._by_id.pop(rev.revision_id, None)
159 return len(revisions)
162# ---------------------------------------------------------------------------
163# RevisionService
164# ---------------------------------------------------------------------------
167class RevisionService:
168 """Manages resource revision snapshots with diff and revert support.
170 Args:
171 store: Revision store implementation. Defaults to
172 :class:`InMemoryRevisionStore`.
173 max_revisions: Maximum revisions retained per record (forwarded to
174 :class:`InMemoryRevisionStore` if no store is provided).
175 """
177 def __init__(
178 self,
179 store: _RevisionStore | None = None,
180 max_revisions: int = 100,
181 ) -> None:
182 self._store: _RevisionStore = store or InMemoryRevisionStore(
183 max_per_record=max_revisions
184 )
185 self._counter: int = 0
187 # ------------------------------------------------------------------
188 # Internal helpers
189 # ------------------------------------------------------------------
191 def _new_id(self) -> str:
192 self._counter += 1
193 ts = datetime.now(UTC).strftime("%Y%m%d%H%M%S%f")
194 return f"rev-{ts}-{self._counter}"
196 # ------------------------------------------------------------------
197 # Public API
198 # ------------------------------------------------------------------
200 async def record(
201 self,
202 resource_type: str,
203 resource_id: str,
204 data: dict[str, Any],
205 *,
206 actor_id: str = "system",
207 comment: str = "",
208 ) -> Revision:
209 """Create and persist a new revision snapshot.
211 Call this *after* a successful save so the snapshot reflects the
212 committed state.
214 Args:
215 resource_type: Resource name (e.g. ``"user"``).
216 resource_id: Record identifier.
217 data: Full field snapshot.
218 actor_id: Principal who triggered the change.
219 comment: Optional human-readable note.
221 Returns:
222 The newly created :class:`Revision`.
223 """
224 revision = Revision(
225 revision_id=self._new_id(),
226 resource_type=resource_type,
227 resource_id=str(resource_id),
228 data=dict(data),
229 actor_id=actor_id,
230 comment=comment,
231 )
232 await self._store.save(revision)
233 return revision
235 async def list_revisions(
236 self,
237 resource_type: str,
238 resource_id: str,
239 *,
240 limit: int = 50,
241 ) -> list[Revision]:
242 """Return revisions for a record, newest-first.
244 Args:
245 resource_type: Resource name.
246 resource_id: Record identifier.
247 limit: Maximum number of revisions to return.
249 Returns:
250 List of :class:`Revision` objects.
251 """
252 return await self._store.list_for_record(
253 resource_type, str(resource_id), limit=limit
254 )
256 async def get_revision(self, revision_id: str) -> Revision | None:
257 """Fetch a single revision by ID.
259 Args:
260 revision_id: The revision identifier.
262 Returns:
263 :class:`Revision` or ``None`` if not found.
264 """
265 return await self._store.get(revision_id)
267 async def diff(
268 self,
269 revision_id_a: str,
270 revision_id_b: str,
271 *,
272 include_unchanged: bool = False,
273 ) -> RevisionDiff | None:
274 """Compute a field-by-field diff between two revisions.
276 Args:
277 revision_id_a: Earlier (or baseline) revision.
278 revision_id_b: Later revision.
279 include_unchanged: When ``True`` all fields are included, not just
280 those that changed.
282 Returns:
283 :class:`RevisionDiff` or ``None`` if either revision is not found.
284 """
285 rev_a = await self._store.get(revision_id_a)
286 rev_b = await self._store.get(revision_id_b)
287 if rev_a is None or rev_b is None:
288 return None
290 all_keys = set(rev_a.data) | set(rev_b.data)
291 diffs: list[FieldDiff] = []
292 for key in sorted(all_keys):
293 old = rev_a.data.get(key)
294 new = rev_b.data.get(key)
295 changed = old != new
296 if include_unchanged or changed:
297 diffs.append(
298 FieldDiff(
299 field_name=key, old_value=old, new_value=new, changed=changed
300 )
301 )
303 return RevisionDiff(
304 from_revision=revision_id_a,
305 to_revision=revision_id_b,
306 fields=diffs,
307 all_fields=include_unchanged,
308 )
310 async def revert_data(self, revision_id: str) -> dict[str, Any] | None:
311 """Return the data snapshot from a prior revision, ready to apply.
313 Does **not** persist anything — callers are responsible for passing
314 the returned data to their update handler (and then calling
315 :meth:`record` to snapshot the revert).
317 Args:
318 revision_id: The revision to revert to.
320 Returns:
321 Field snapshot dict or ``None`` if revision not found.
322 """
323 revision = await self._store.get(revision_id)
324 return dict(revision.data) if revision else None
326 async def purge(self, resource_type: str, resource_id: str) -> int:
327 """Delete all revisions for a record.
329 Args:
330 resource_type: Resource name.
331 resource_id: Record identifier.
333 Returns:
334 Number of revisions deleted.
335 """
336 return await self._store.delete_for_record(resource_type, str(resource_id))
339__all__ = [
340 "FieldDiff",
341 "InMemoryRevisionStore",
342 "Revision",
343 "RevisionDiff",
344 "RevisionService",
345]