Skip to content

RevisionAbstractSkel

RevisionAbstractSkel

Bases: Skeleton

Skeleton mixin wiring in the RevisionDatabaseAdapter.

revision_fields selects the bones that trigger a revision; ["*"] (default) tracks every write.

Source code in src/viur/revision/skel.py 
 9
10
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
class RevisionAbstractSkel(Skeleton):
    """Skeleton mixin wiring in the
    [`RevisionDatabaseAdapter`][viur.revision.adapter.RevisionDatabaseAdapter].

    ``revision_fields`` selects the bones that trigger a revision; ``["*"]``
    (default) tracks every write.
    """

    revision_fields: t.ClassVar[list[str]] = ["*"]

    def __new__(cls, *args, **kwargs):
        instance = super().__new__(cls, *args, **kwargs)
        # Some bases reset database_adapters in __new__; re-append ours if gone.
        adapters = list(getattr(cls, "database_adapters", []))
        if not any(isinstance(a, RevisionDatabaseAdapter) for a in adapters):
            adapters.append(RevisionDatabaseAdapter())
            cls.database_adapters = adapters
        return instance

    @classmethod
    def read_revision(
        cls,
        skel: SkeletonInstance,
        key: t.Union["db.Key", str],
        version: t.Union[str, int] = "latest",
    ) -> t.Optional[SkeletonInstance]:
        """Populate *skel* with a revision of *key* (``db.Key`` or legacy-urlsafe).

        Same convention as ``Skeleton.read``. *version* is ``"latest"`` or a
        revision index. Returns the skel, or ``None`` if not found.
        """
        if isinstance(key, str):
            try:
                key = db.Key.from_legacy_urlsafe(key)
            except Exception:
                return None

        revision_kind = f"{skel.kindName}_revision"
        query = db.Query(revision_kind).filter("origin_key =", key)

        if version == "latest":
            entry = query.order(("revision_index", db.SortOrder.Descending)).getEntry()
        else:
            try:
                version = int(version)
            except (TypeError, ValueError):
                return None
            entry = query.filter("revision_index =", version).getEntry()

        if entry is None:
            return None

        skel.setEntity(entry)
        return skel

read_revision classmethod 

read_revision(skel: SkeletonInstance, key: Union[Key, str], version: Union[str, int] = 'latest') -> Optional[SkeletonInstance]

Populate skel with a revision of key (db.Key or legacy-urlsafe).

Same convention as Skeleton.read. version is "latest" or a revision index. Returns the skel, or None if not found.

Source code in src/viur/revision/skel.py 
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
@classmethod
def read_revision(
    cls,
    skel: SkeletonInstance,
    key: t.Union["db.Key", str],
    version: t.Union[str, int] = "latest",
) -> t.Optional[SkeletonInstance]:
    """Populate *skel* with a revision of *key* (``db.Key`` or legacy-urlsafe).

    Same convention as ``Skeleton.read``. *version* is ``"latest"`` or a
    revision index. Returns the skel, or ``None`` if not found.
    """
    if isinstance(key, str):
        try:
            key = db.Key.from_legacy_urlsafe(key)
        except Exception:
            return None

    revision_kind = f"{skel.kindName}_revision"
    query = db.Query(revision_kind).filter("origin_key =", key)

    if version == "latest":
        entry = query.order(("revision_index", db.SortOrder.Descending)).getEntry()
    else:
        try:
            version = int(version)
        except (TypeError, ValueError):
            return None
        entry = query.filter("revision_index =", version).getEntry()

    if entry is None:
        return None

    skel.setEntity(entry)
    return skel