RevisionModule

RevisionModule

Revision list/view/apply/delete endpoints plus daily retention.

Every endpoint requires root or <moduleName>-manage. Pair with a RevisionAbstractSkel-based skel.

Source code in src/viur/revision/module.py

class RevisionModule:
    """Revision list/view/apply/delete endpoints plus daily retention.

    Every endpoint requires ``root`` or ``<moduleName>-manage``. Pair with a
    [`RevisionAbstractSkel`][viur.revision.skel.RevisionAbstractSkel]-based skel.
    """

    def _require_manage_access(self) -> None:
        cuser = current.user.get()
        if not cuser:
            raise errors.Unauthorized()
        access = cuser.get("access") or []
        if "root" not in access and f"{self.moduleName}-manage" not in access:
            raise errors.Forbidden()

    @staticmethod
    def _parse_key(key: str) -> "db.Key":
        try:
            return db.Key.from_legacy_urlsafe(key)
        except Exception:
            raise errors.BadRequest("Ungültiger Key")

    @staticmethod
    def _parse_version(version: t.Union[str, int]) -> t.Union[str, int]:
        if version == "latest":
            return "latest"
        try:
            return int(version)
        except (TypeError, ValueError):
            raise errors.BadRequest("Ungültige Versionsnummer")

    # ------------------------------------------------------------------ #
    # Public endpoints                                                     #
    # ------------------------------------------------------------------ #

    @exposed
    def list_revision(self, key: str, *args, **kwargs):
        """List revisions for *key* (legacy-urlsafe), newest first.

        Standard ``list`` render over the ``{kind}_revision`` kind. Each skel's
        ``changedate`` is overwritten with the snapshot's ``revision_date`` so
        the UI shows when the revision was captured.
        """
        self._require_manage_access()
        origin_key = self._parse_key(key)

        query = self.viewSkel().all()
        revision_kind = f"{query.srcSkel.kindName}_revision"
        query.kind = revision_kind
        query.queries.kind = revision_kind

        query.filter("origin_key =", origin_key)
        query.order(("revision_index", db.SortOrder.Descending))

        skellist = query.fetch(100)
        for skel in skellist:
            rev_date = skel.dbEntity.get("revision_date") if skel.dbEntity else None
            if rev_date is not None:
                skel["changedate"] = rev_date

        return self.render.list(skellist)

    @exposed
    def view_revision(
        self,
        key: str = None,
        version: t.Union[str, int] = "latest",
        revision_key: str = None,
        *args,
        **kwargs,
    ):
        """Render a single revision via the module's ``view`` renderer.

        :param revision_key: Key in ``{kind}_revision`` (preferred).
        :param key: Original entity key; used with *version* if no *revision_key*.
        :param version: ``"latest"`` or a revision index.
        """
        self._require_manage_access()
        skel = self.viewSkel()

        if revision_key:
            entry = db.Get(self._parse_key(revision_key))
            if entry is None:
                raise errors.NotFound("Revision nicht gefunden")
            skel.setEntity(entry)
        else:
            if not skel.read_revision(self._parse_key(key), self._parse_version(version)):
                raise errors.NotFound("Revision nicht gefunden")

        return self.render.view(skel)

    @exposed
    @force_post
    @skey
    def apply_revision(
        self,
        key: str = None,
        version: t.Union[str, int] = "latest",
        revision_key: str = None,
        *args,
        **kwargs,
    ) -> str:
        """Restore the live entity to a revision.

        The current live state is first snapshotted (tagged
        ``revision_change_list=["*restore*"]`` + ``restore_from_index``), so the
        restore is itself undoable.

        :param revision_key: Key in ``{kind}_revision`` (preferred).
        :param key: Original entity key; used with *version* if no *revision_key*.
        :param version: ``"latest"`` or a revision index.
        :returns: JSON ``{"status": "ok", "restored_revision": <index>}``.
        """
        self._require_manage_access()
        revision_kind = f"{self.viewSkel().kindName}_revision"

        if revision_key:
            target_entry = db.Get(self._parse_key(revision_key))
            if target_entry is None:
                raise errors.NotFound("Revision nicht gefunden")
            origin_key = target_entry.get("origin_key")
            if origin_key is None:
                raise errors.BadRequest("Revision enthält keinen origin_key")
        else:
            origin_key = self._parse_key(key)
            skel = self.viewSkel()
            if not skel.read_revision(origin_key, self._parse_version(version)):
                raise errors.NotFound("Revision nicht gefunden")
            target_entry = skel.dbEntity

        restored_index = target_entry.get("revision_index")

        # --- 1. Snapshot the current live state before overwriting ----------
        live_entity = db.Get(origin_key)
        if live_entity is not None:
            existing_latest = (
                db.Query(revision_kind)
                .filter("origin_key =", origin_key)
                .order(("revision_index", db.SortOrder.Descending))
                .getEntry()
            )
            next_index = (existing_latest["revision_index"] + 1) if existing_latest else 1

            snapshot_key = db.AllocateIDs(db.Key(revision_kind))
            snapshot_entity = db.Entity(snapshot_key)
            for k, v in live_entity.items():
                snapshot_entity[k] = v
            snapshot_entity["origin_key"] = origin_key
            snapshot_entity["revision_index"] = next_index
            snapshot_entity["revision_date"] = utils.utcNow()
            snapshot_entity["revision_change_list"] = ["*restore*"]
            snapshot_entity["restore_from_index"] = restored_index
            db.Put(snapshot_entity)

        # --- 2. Restore the target revision into the live entity ------------
        restore_entity = db.Entity(origin_key)
        for k, v in target_entry.items():
            if k not in _REVISION_META_KEYS:
                restore_entity[k] = v
        db.Put(restore_entity)

        logger.info(
            "apply_revision: %s restored to revision %s by %s",
            origin_key,
            restored_index,
            (current.user.get() or {}).get("name", "unknown"),
        )

        return json.dumps(
            {"status": "ok", "restored_revision": restored_index},
            cls=CustomJsonEncoder,
        )

    @exposed
    @force_post
    @skey
    def delete_revision(
        self,
        revision_key: str,
        *args,
        **kwargs,
    ) -> str:
        """Delete a single revision snapshot.

        *revision_key* must belong to this module's ``{kind}_revision`` kind;
        live entity keys are rejected.

        :returns: JSON ``{"status": "ok", "deleted_revision": <index>}``.
        """
        self._require_manage_access()
        revision_kind = f"{self.viewSkel().kindName}_revision"

        key = self._parse_key(revision_key)
        if key.kind != revision_kind:
            raise errors.BadRequest("Key gehört nicht zu einer Revision dieses Moduls")

        entry = db.Get(key)
        if entry is None:
            raise errors.NotFound("Revision nicht gefunden")

        deleted_index = entry.get("revision_index")
        db.Delete(key)

        logger.info(
            "delete_revision: %s (index %s) deleted by %s",
            key,
            deleted_index,
            (current.user.get() or {}).get("name", "unknown"),
        )

        return json.dumps(
            {"status": "ok", "deleted_revision": deleted_index},
            cls=CustomJsonEncoder,
        )

    # ------------------------------------------------------------------ #
    # Retention                                                            #
    # ------------------------------------------------------------------ #

    @PeriodicTask(datetime.timedelta(days=1), cronName="default")
    def cleanup_revisions(self, *args, **kwargs):
        """Daily retention per ``origin_key`` (latest per slot survives):
        <24h keep all, 1–7d 1/hour, 7–30d 1/day, older 1/week.
        """
        return self._run_cleanup_revisions()

    @exposed
    def cleanup_revisions_now(self, *args, **kwargs) -> str:
        """Run retention on demand, returning per-revision diagnostics
        (bucket key, age, kept/deleted) alongside ``status``/``deleted``/``origins``.
        """
        self._require_manage_access()
        deleted_total, origins_total, entries, now, revision_kind = self._run_cleanup_revisions(
            collect_entries=True
        )
        return json.dumps(
            {
                "status": "ok",
                "deleted": deleted_total,
                "origins": origins_total,
                "now": str(now),
                "now_tzinfo": str(now.tzinfo),
                "revision_kind": revision_kind,
                "count": len(entries),
                "entries": entries,
            },
            cls=CustomJsonEncoder,
        )

    def _run_cleanup_revisions(self, collect_entries: bool = False):
        """Returns ``(deleted_total, origins_total)``, or, with
        *collect_entries*, ``(..., entries, now, revision_kind)``.
        """
        revision_kind = f"{self.viewSkel().kindName}_revision"
        now = utils.utcNow()

        # Loads all revisions into memory; split via an origin_key cursor if the
        # dataset outgrows a single run.
        by_origin: dict = {}
        for rev in db.Query(revision_kind).iter():
            origin = rev.get("origin_key")
            if origin is not None:
                by_origin.setdefault(origin, []).append(rev)

        diagnostics: list = [] if collect_entries else None
        deleted_total = 0
        for revs in by_origin.values():
            deleted_total += self._thin_origin_revisions(revs, now, diagnostics)

        logger.info(
            "cleanup_revisions(%s): %d revisions deleted across %d origins",
            revision_kind,
            deleted_total,
            len(by_origin),
        )

        if collect_entries:
            return deleted_total, len(by_origin), diagnostics, now, revision_kind
        return deleted_total, len(by_origin)

    @staticmethod
    def _thin_origin_revisions(revs: list, now: datetime.datetime, diagnostics: list = None) -> int:
        """Apply bucket retention to one origin's revisions; returns the delete
        count. If *diagnostics* is a list, per-revision metadata is appended."""
        if len(revs) <= 1:
            if diagnostics is not None:
                for rev in revs:
                    diagnostics.append({
                        "key": str(rev.key),
                        "revision_index": rev.get("revision_index"),
                        "revision_date": str(rev.get("revision_date")),
                        "origin_key": str(rev.get("origin_key")),
                        "reason": "only-rev-for-origin",
                        "action": "keep",
                    })
            return 0

        bucket_winners: dict = {}  # bucket_key -> rev with newest revision_date
        keep_always: list = []
        per_rev_info: dict = {}    # rev.key -> diagnostic dict (filled progressively)

        for rev in revs:
            info = {
                "key": str(rev.key),
                "revision_index": rev.get("revision_index"),
                "revision_date": str(rev.get("revision_date")),
                "rev_date_tzinfo": str(getattr(rev.get("revision_date"), "tzinfo", None)),
                "origin_key": str(rev.get("origin_key")),
            }
            per_rev_info[rev.key] = info

            rev_date = rev.get("revision_date")
            if not rev_date:
                info["reason"] = "no-revision_date"
                keep_always.append(rev)
                continue

            try:
                age = now - rev_date
            except Exception as e:
                info["error"] = repr(e)
                keep_always.append(rev)
                continue

            info["age"] = str(age)
            bucket_fn = RevisionModule._bucket_for_age(age)
            if bucket_fn is None:
                info["reason"] = "within-keep-all-window"
                keep_always.append(rev)
                continue

            bucket_key = bucket_fn(rev_date)
            info["bucket_key"] = str(bucket_key)
            prev = bucket_winners.get(bucket_key)
            if prev is None or rev_date > prev["revision_date"]:
                bucket_winners[bucket_key] = rev

        keep_keys = {r.key for r in keep_always}
        keep_keys.update(r.key for r in bucket_winners.values())

        deleted = 0
        for rev in revs:
            info = per_rev_info.get(rev.key)
            if rev.key not in keep_keys:
                db.Delete(rev.key)
                deleted += 1
                if info is not None:  # pragma: no branch - per_rev_info always populated
                    info["action"] = "delete"
            elif info is not None and "action" not in info:  # pragma: no branch
                info["action"] = "keep"

        if diagnostics is not None:
            diagnostics.extend(per_rev_info.values())
        return deleted

    @staticmethod
    def _bucket_for_age(age: datetime.timedelta):
        """Return the bucket function for ``age`` or ``None`` to keep all."""
        for max_age, bucket_fn in _RETENTION_BUCKETS:
            if max_age is None or age < max_age:
                return bucket_fn
        return None  # pragma: no cover - last bucket has max_age=None, always matches

list_revision

list_revision(key: str, *args, **kwargs)

List revisions for key (legacy-urlsafe), newest first.

Standard list render over the {kind}_revision kind. Each skel's changedate is overwritten with the snapshot's revision_date so the UI shows when the revision was captured.

Source code in src/viur/revision/module.py

@exposed
def list_revision(self, key: str, *args, **kwargs):
    """List revisions for *key* (legacy-urlsafe), newest first.

    Standard ``list`` render over the ``{kind}_revision`` kind. Each skel's
    ``changedate`` is overwritten with the snapshot's ``revision_date`` so
    the UI shows when the revision was captured.
    """
    self._require_manage_access()
    origin_key = self._parse_key(key)

    query = self.viewSkel().all()
    revision_kind = f"{query.srcSkel.kindName}_revision"
    query.kind = revision_kind
    query.queries.kind = revision_kind

    query.filter("origin_key =", origin_key)
    query.order(("revision_index", db.SortOrder.Descending))

    skellist = query.fetch(100)
    for skel in skellist:
        rev_date = skel.dbEntity.get("revision_date") if skel.dbEntity else None
        if rev_date is not None:
            skel["changedate"] = rev_date

    return self.render.list(skellist)

view_revision

view_revision(key: str = None, version: Union[str, int] = 'latest', revision_key: str = None, *args, **kwargs)

Render a single revision via the module's view renderer.

Parameters:

Name	Type	Description	Default
revision_key	str	Key in {kind}_revision (preferred).	None
key	str	Original entity key; used with version if no revision_key.	None
version	Union[str, int]	"latest" or a revision index.	'latest'

Source code in src/viur/revision/module.py

@exposed
def view_revision(
    self,
    key: str = None,
    version: t.Union[str, int] = "latest",
    revision_key: str = None,
    *args,
    **kwargs,
):
    """Render a single revision via the module's ``view`` renderer.

    :param revision_key: Key in ``{kind}_revision`` (preferred).
    :param key: Original entity key; used with *version* if no *revision_key*.
    :param version: ``"latest"`` or a revision index.
    """
    self._require_manage_access()
    skel = self.viewSkel()

    if revision_key:
        entry = db.Get(self._parse_key(revision_key))
        if entry is None:
            raise errors.NotFound("Revision nicht gefunden")
        skel.setEntity(entry)
    else:
        if not skel.read_revision(self._parse_key(key), self._parse_version(version)):
            raise errors.NotFound("Revision nicht gefunden")

    return self.render.view(skel)

apply_revision

apply_revision(key: str = None, version: Union[str, int] = 'latest', revision_key: str = None, *args, **kwargs) -> str

Restore the live entity to a revision.

The current live state is first snapshotted (tagged revision_change_list=["*restore*"] + restore_from_index), so the restore is itself undoable.

Parameters:

Name	Type	Description	Default
revision_key	str	Key in {kind}_revision (preferred).	None
key	str	Original entity key; used with version if no revision_key.	None
version	Union[str, int]	"latest" or a revision index.	'latest'

Returns:

Type	Description
str	JSON {"status": "ok", "restored_revision": <index>}.

Source code in src/viur/revision/module.py

@exposed
@force_post
@skey
def apply_revision(
    self,
    key: str = None,
    version: t.Union[str, int] = "latest",
    revision_key: str = None,
    *args,
    **kwargs,
) -> str:
    """Restore the live entity to a revision.

    The current live state is first snapshotted (tagged
    ``revision_change_list=["*restore*"]`` + ``restore_from_index``), so the
    restore is itself undoable.

    :param revision_key: Key in ``{kind}_revision`` (preferred).
    :param key: Original entity key; used with *version* if no *revision_key*.
    :param version: ``"latest"`` or a revision index.
    :returns: JSON ``{"status": "ok", "restored_revision": <index>}``.
    """
    self._require_manage_access()
    revision_kind = f"{self.viewSkel().kindName}_revision"

    if revision_key:
        target_entry = db.Get(self._parse_key(revision_key))
        if target_entry is None:
            raise errors.NotFound("Revision nicht gefunden")
        origin_key = target_entry.get("origin_key")
        if origin_key is None:
            raise errors.BadRequest("Revision enthält keinen origin_key")
    else:
        origin_key = self._parse_key(key)
        skel = self.viewSkel()
        if not skel.read_revision(origin_key, self._parse_version(version)):
            raise errors.NotFound("Revision nicht gefunden")
        target_entry = skel.dbEntity

    restored_index = target_entry.get("revision_index")

    # --- 1. Snapshot the current live state before overwriting ----------
    live_entity = db.Get(origin_key)
    if live_entity is not None:
        existing_latest = (
            db.Query(revision_kind)
            .filter("origin_key =", origin_key)
            .order(("revision_index", db.SortOrder.Descending))
            .getEntry()
        )
        next_index = (existing_latest["revision_index"] + 1) if existing_latest else 1

        snapshot_key = db.AllocateIDs(db.Key(revision_kind))
        snapshot_entity = db.Entity(snapshot_key)
        for k, v in live_entity.items():
            snapshot_entity[k] = v
        snapshot_entity["origin_key"] = origin_key
        snapshot_entity["revision_index"] = next_index
        snapshot_entity["revision_date"] = utils.utcNow()
        snapshot_entity["revision_change_list"] = ["*restore*"]
        snapshot_entity["restore_from_index"] = restored_index
        db.Put(snapshot_entity)

    # --- 2. Restore the target revision into the live entity ------------
    restore_entity = db.Entity(origin_key)
    for k, v in target_entry.items():
        if k not in _REVISION_META_KEYS:
            restore_entity[k] = v
    db.Put(restore_entity)

    logger.info(
        "apply_revision: %s restored to revision %s by %s",
        origin_key,
        restored_index,
        (current.user.get() or {}).get("name", "unknown"),
    )

    return json.dumps(
        {"status": "ok", "restored_revision": restored_index},
        cls=CustomJsonEncoder,
    )

delete_revision

delete_revision(revision_key: str, *args, **kwargs) -> str

Delete a single revision snapshot.

revision_key must belong to this module's {kind}_revision kind; live entity keys are rejected.

Returns:

Type	Description
str	JSON {"status": "ok", "deleted_revision": <index>}.

Source code in src/viur/revision/module.py

@exposed
@force_post
@skey
def delete_revision(
    self,
    revision_key: str,
    *args,
    **kwargs,
) -> str:
    """Delete a single revision snapshot.

    *revision_key* must belong to this module's ``{kind}_revision`` kind;
    live entity keys are rejected.

    :returns: JSON ``{"status": "ok", "deleted_revision": <index>}``.
    """
    self._require_manage_access()
    revision_kind = f"{self.viewSkel().kindName}_revision"

    key = self._parse_key(revision_key)
    if key.kind != revision_kind:
        raise errors.BadRequest("Key gehört nicht zu einer Revision dieses Moduls")

    entry = db.Get(key)
    if entry is None:
        raise errors.NotFound("Revision nicht gefunden")

    deleted_index = entry.get("revision_index")
    db.Delete(key)

    logger.info(
        "delete_revision: %s (index %s) deleted by %s",
        key,
        deleted_index,
        (current.user.get() or {}).get("name", "unknown"),
    )

    return json.dumps(
        {"status": "ok", "deleted_revision": deleted_index},
        cls=CustomJsonEncoder,
    )

cleanup_revisions

cleanup_revisions(*args, **kwargs)

Daily retention per origin_key (latest per slot survives): <24h keep all, 1–7d 1/hour, 7–30d 1/day, older 1/week.

Source code in src/viur/revision/module.py

@PeriodicTask(datetime.timedelta(days=1), cronName="default")
def cleanup_revisions(self, *args, **kwargs):
    """Daily retention per ``origin_key`` (latest per slot survives):
    <24h keep all, 1–7d 1/hour, 7–30d 1/day, older 1/week.
    """
    return self._run_cleanup_revisions()

cleanup_revisions_now

cleanup_revisions_now(*args, **kwargs) -> str

Run retention on demand, returning per-revision diagnostics (bucket key, age, kept/deleted) alongside status/deleted/origins.

Source code in src/viur/revision/module.py

@exposed
def cleanup_revisions_now(self, *args, **kwargs) -> str:
    """Run retention on demand, returning per-revision diagnostics
    (bucket key, age, kept/deleted) alongside ``status``/``deleted``/``origins``.
    """
    self._require_manage_access()
    deleted_total, origins_total, entries, now, revision_kind = self._run_cleanup_revisions(
        collect_entries=True
    )
    return json.dumps(
        {
            "status": "ok",
            "deleted": deleted_total,
            "origins": origins_total,
            "now": str(now),
            "now_tzinfo": str(now.tzinfo),
            "revision_kind": revision_kind,
            "count": len(entries),
            "entries": entries,
        },
        cls=CustomJsonEncoder,
    )