popoto.fields.access_tracker

popoto.fields.access_tracker

AccessTrackerMixin — read pattern tracking with staged vs confirmed reads.

This module provides a mixin class that adds read access tracking to any Model. It uses a staging pattern: reads are first recorded to a staging list, then atomically promoted to the confirmed access log via a Lua script.

Design

• on_read() appends a timestamp to a staging list (cheap, fire-and-forget)
• confirm_access() atomically promotes staged timestamps to the confirmed log
• discard_staged_access() discards staged reads without affecting confirmed data
• Access log is capped at max_access_log entries (default 100)

Redis Key Patterns

• $AT:{ClassName}:staged:{redis_key} — staging list (RPUSH timestamps)
• $AT:{ClassName}:access_log:{redis_key} — confirmed access timestamps
• $AT:{ClassName}:meta:{redis_key} — hash with access_count and last_accessed

Example

class Memory(AccessTrackerMixin, Model): key = UniqueKeyField() content = StringField()

memory = Memory.query.get(key="important") # auto-stages on_read memory.confirm_access() # promote staged reads to confirmed log print(memory.access_count) # total confirmed read count print(memory.last_accessed) # timestamp of most recent confirmed read

AccessTrackerMixin

Mixin that adds read access tracking to any Model.

Add this as a base class alongside Model to enable read tracking:

class MyModel(AccessTrackerMixin, Model):
    ...

Class Attributes

_max_access_log: Maximum number of timestamps to keep in the access log. Older entries are trimmed on confirm. Default 100. _track_reads: Whether to automatically track reads from queries. Default True. _staged_ttl_seconds: TTL applied to the staging list on every on_read() call. Default 86400 (24h). Magic-number tuning knob — increase if your stage→confirm cadence exceeds 24h. See on_read() docstring for the staged-read TTL contract.

Note: Attributes are prefixed with underscore to avoid conflict with Popoto's ModelBase metaclass, which requires public attributes to be Fields.

Source code in src/popoto/fields/access_tracker.py

class AccessTrackerMixin:
    """Mixin that adds read access tracking to any Model.

    Add this as a base class alongside Model to enable read tracking:

        class MyModel(AccessTrackerMixin, Model):
            ...

    Class Attributes:
        _max_access_log: Maximum number of timestamps to keep in the access log.
            Older entries are trimmed on confirm. Default 100.
        _track_reads: Whether to automatically track reads from queries.
            Default True.
        _staged_ttl_seconds: TTL applied to the staging list on every on_read()
            call. Default 86400 (24h). Magic-number tuning knob — increase if
            your stage→confirm cadence exceeds 24h. See on_read() docstring for
            the staged-read TTL contract.

    Note: Attributes are prefixed with underscore to avoid conflict with
    Popoto's ModelBase metaclass, which requires public attributes to be Fields.
    """

    _max_access_log = 100
    _track_reads = True
    _staged_ttl_seconds: int = (
        86400  # 24h — magic-number tuning knob; see on_read() docstring
    )

    # Export/import: this is a Model-level mixin, reached by the transfer
    # driver's MRO walk rather than by iterating _meta.fields, so its
    # export_state/import_state take the same signature minus ``field_name``.
    # The meta hash counters and the confirmed access log are carried: both
    # are per-record facts about reads that genuinely happened on the source.
    # The staged list is deliberately NOT carried, and that is why this stays
    # "partial" -- staged entries are uncommitted, TTL-bounded reads awaiting
    # confirm_access(), so dropping them is observably identical to calling
    # discard_staged_access(), a first-class operation. An in-flight,
    # unconfirmed read not surviving a transfer is correct, not a gap. #556
    # settled this as a permanent contract rather than pending work.
    roundtrip_policy: str = "partial"
    roundtrip_note: str = (
        "access_count, last_accessed, and the confirmed access log "
        "($AT:{Class}:access_log:{key}) are carried. Staged (unconfirmed, "
        "TTL-bounded) reads are not: dropping an in-flight read is equivalent "
        "to discard_staged_access(). This is a permanent contract, not "
        "pending work."
    )

    @classmethod
    def export_state(cls, model_instance, **kwargs):
        """Export the access-tracker meta counters and confirmed log.

        Returns:
            ``{"access_count": int, "last_accessed": float,
            "access_log": [float, ...]}``, or ``None`` when this instance has
            never had a confirmed access. Each key is omitted when its source
            structure is absent, so an instance with counters but an empty log
            exports exactly the shape #554-era code produced.
        """
        meta_key = model_instance._at_key("meta")
        raw_count = get_REDIS_DB().hget(meta_key, "access_count")
        raw_last = get_REDIS_DB().hget(meta_key, "last_accessed")
        raw_log = get_REDIS_DB().lrange(model_instance._at_key("access_log"), 0, -1)
        if raw_count is None and raw_last is None and not raw_log:
            return None

        state: dict[str, object] = {}
        if raw_count is not None:
            try:
                state["access_count"] = int(raw_count)
            except (TypeError, ValueError):
                logger.warning(f"Non-numeric access_count in {meta_key}; skipping")
        if raw_last is not None:
            try:
                state["last_accessed"] = float(raw_last)
            except (TypeError, ValueError):
                logger.warning(f"Non-numeric last_accessed in {meta_key}; skipping")
        if raw_log:
            # Decoded to floats here rather than left as bytes: to_jsonable
            # tags bytes with a __bytes__ envelope, which round-trips but
            # stops matching the plain-list shape the docs describe.
            log: list[float] = []
            for raw_ts in raw_log:
                try:
                    log.append(float(raw_ts))
                except (TypeError, ValueError):
                    logger.warning(
                        f"Non-numeric access_log entry for {meta_key}; skipping entry"
                    )
            if log:
                state["access_log"] = log
        return state or None

    @classmethod
    def import_state(cls, model_instance, state, **kwargs):
        """Restore the access-tracker meta counters and confirmed log.

        ``access_count`` and ``last_accessed`` are exposed only as read-only
        properties, so this writes the meta hash directly -- the same hash
        ``CONFIRM_ACCESS_LUA`` maintains. The log is restored by a raw
        DELETE + RPUSH rather than by staging the timestamps and calling
        ``confirm_access()``: that path would HINCRBY the count a second time
        on top of the carried value and overwrite ``last_accessed``, and it
        would push the timestamps through a staging list whose TTL contract
        has nothing to do with a transfer.

        The log is trimmed to the *destination's* ``_max_access_log``, keeping
        the most recent entries -- the same trim ``CONFIRM_ACCESS_LUA``
        applies at ``:47``. ``access_log`` absent from ``state`` is tolerated:
        a file exported by #554-era code has no such key, and must import
        cleanly rather than raise.
        """
        if not state:
            return None

        mapping: dict[str, Any] = {}
        if state.get("access_count") is not None:
            mapping["access_count"] = int(state["access_count"])
        if state.get("last_accessed") is not None:
            mapping["last_accessed"] = float(state["last_accessed"])
        if mapping:
            get_REDIS_DB().hset(model_instance._at_key("meta"), mapping=mapping)

        log = state.get("access_log")
        if isinstance(log, (list, tuple)) and log:
            cap = max(1, int(cls._max_access_log))
            entries = [str(float(ts)) for ts in log][-cap:]
            log_key = model_instance._at_key("access_log")
            get_REDIS_DB().delete(log_key)
            get_REDIS_DB().rpush(log_key, *entries)
        return None

    def _at_key(self, kind):
        """Build an access tracker Redis key.

        Args:
            kind: One of 'staged', 'access_log', 'meta'

        Returns:
            str: Redis key like '$AT:ClassName:kind:redis_key'
        """
        class_name = type(self).__name__
        redis_key = self._redis_key or self.db_key.redis_key
        return f"$AT:{class_name}:{kind}:{redis_key}"

    def on_read(self, pipeline=None):
        """Record a read access by staging a timestamp.

        Appends the current timestamp to the staging list and refreshes the
        TTL on the staged key. This is a cheap operation suitable for
        fire-and-forget use from query hooks.

        Staged-read TTL contract:
            Applications MUST call ``confirm_access()`` within
            ``_staged_ttl_seconds`` (default 24h) of staging reads via
            ``on_read()``. Reads left unconfirmed past the TTL window are
            permanently dropped from ``access_count`` / ``last_accessed`` by
            design. Set ``_staged_ttl_seconds`` higher if a longer
            stage→confirm cadence is required. The ``_staged_ttl_seconds``
            constant is a magic-number tuning knob — it is not user-facing
            configuration.

        Args:
            pipeline: Optional Redis pipeline for batch operations. When
                provided, the RPUSH and EXPIRE are queued in the same pipeline
                call so they execute atomically.
        """
        backend = non_redis_backend(self)
        if backend is not None:
            # #759 M2a: _staged_reads/_staged_at on the record's own row.
            self._access_call(
                backend,
                "stage",
                {self._at_member(): 1},
                now=time.time(),
                ttl=self._staged_ttl_seconds,
                uow=_uow_of(pipeline, backend),
            )
            return
        ts = str(time.time())
        staged_key = self._at_key("staged")
        if pipeline:
            pipeline.rpush(staged_key, ts)
            pipeline.expire(staged_key, self._staged_ttl_seconds)
        else:
            get_REDIS_DB().rpush(staged_key, ts)
            get_REDIS_DB().expire(staged_key, self._staged_ttl_seconds)

    def confirm_access(self, pipeline=None):
        """Atomically promote staged reads to the confirmed access log.

        Uses a Lua script to:
        1. Move all staged timestamps to the access log
        2. Trim the log to max_access_log entries
        3. Update access_count and last_accessed in the meta hash
        4. Delete the staging list

        TTL contract (confirm side):
            A staged key that expired before confirm contributes 0 to
            ``access_count`` and does not update ``last_accessed`` — this is
            by design (see on_read() staged-read TTL contract). When this
            happens a DEBUG log is emitted.

        Args:
            pipeline: Optional Redis pipeline (not used for Lua eval,
                reserved for future use).

        Returns:
            int: Number of staged reads that were promoted (0 if the staged
                key was empty or had already expired).

        Raises:
            TypeError: If the model instance has not been saved to Redis.
        """
        if not hasattr(self, "_redis_key") and not hasattr(self, "db_key"):
            raise TypeError("confirm_access() requires a saved model instance")
        try:
            redis_key = self._redis_key or self.db_key.redis_key
        except Exception:
            raise TypeError("confirm_access() requires a saved model instance")

        backend = non_redis_backend(self)
        if backend is not None:
            # #759 M2a: one UPDATE promotes the live staged reads.
            promoted = self._access_call(
                backend,
                "confirm",
                self._at_record_id(),
                now=time.time(),
                ttl=self._staged_ttl_seconds,
                uow=_uow_of(pipeline, backend),
            )
            if promoted is None:
                raise TypeError("confirm_access() requires a saved model instance")
            if promoted == 0:
                logger.debug(
                    "AccessTracker: staged key empty/expired at confirm for %s — "
                    "read dropped (TTL contract)",
                    self._at_member(),
                )
            return promoted

        # Check if the model has been saved (has a valid redis key in the DB)
        if not get_REDIS_DB().exists(redis_key):
            raise TypeError("confirm_access() requires a saved model instance")

        staged_key = self._at_key("staged")
        log_key = self._at_key("access_log")
        meta_key = self._at_key("meta")

        count = run_lua(
            get_REDIS_DB(),
            CONFIRM_ACCESS_LUA,
            3,  # number of KEYS
            staged_key,
            log_key,
            meta_key,
            str(self._max_access_log),
        )
        count = int(count)
        if count == 0:
            logger.debug(
                "AccessTracker: staged key empty/expired at confirm for %s — read dropped (TTL contract)",
                self.db_key,
            )
        return count

    def discard_staged_access(self, pipeline=None):
        """Discard all staged reads without affecting confirmed data.

        Args:
            pipeline: Optional Redis pipeline for batch operations.
        """
        backend = non_redis_backend(self)
        if backend is not None:
            self._access_call(
                backend, "discard", self._at_record_id(), uow=_uow_of(pipeline, backend)
            )
            return
        staged_key = self._at_key("staged")
        if pipeline:
            pipeline.delete(staged_key)
        else:
            get_REDIS_DB().delete(staged_key)

    @property
    def access_count(self):
        """Total number of confirmed read accesses.

        Returns:
            int: The cumulative access count, or 0 if never confirmed.
        """
        backend = non_redis_backend(self)
        if backend is not None:
            return self._access_state(backend)[0]
        meta_key = self._at_key("meta")
        raw = get_REDIS_DB().hget(meta_key, "access_count")
        if raw is None:
            return 0
        return int(raw)

    @property
    def last_accessed(self):
        """Timestamp of the most recent confirmed read access.

        Returns:
            float or None: Unix timestamp, or None if never confirmed.
        """
        backend = non_redis_backend(self)
        if backend is not None:
            return self._access_state(backend)[1]
        meta_key = self._at_key("meta")
        raw = get_REDIS_DB().hget(meta_key, "last_accessed")
        if raw is None:
            return None
        return float(raw)

    # -- non-Redis backends (#759 M2a) -------------------------------------

    def _at_member(self) -> Any:
        """The key this instance's tracking is filed under (as _at_key)."""
        model: Any = self
        return model._redis_key or model.db_key.redis_key

    def _at_record_id(self) -> Any:
        from ..backends import RecordId

        model: Any = self
        return RecordId.from_key(model._meta.model_name, self._at_member())

    def _access_call(self, backend: Any, op: str, *args: Any, **kwargs: Any) -> Any:
        model: Any = self
        return backend.field_call(model._meta.spec, "_access", op, *args, **kwargs)

    def _access_state(self, backend: Any) -> Any:
        """``(access_count, last_accessed, live staged reads)`` from the
        record's row; a missing record reads as never accessed."""
        return self._access_call(
            backend,
            "state",
            self._at_record_id(),
            now=time.time(),
            ttl=self._staged_ttl_seconds,
        )

    @classmethod
    def _stage_reads_on_backend(cls, backend: Any, instances: Any) -> None:
        """Stage one read per instance in one statement (the bulk
        ``_fire_on_read`` path), rows locked in ``_pk`` order."""
        counts: dict[str, int] = {}
        for inst in instances:
            key = inst._at_member()
            counts[key] = counts.get(key, 0) + 1
        if counts:
            model: Any = cls
            backend.field_call(
                model._meta.spec,
                "_access",
                "stage",
                counts,
                now=time.time(),
                ttl=cls._staged_ttl_seconds,
            )

    def _delete_access_tracker_keys(self, pipeline=None):
        """Remove all access tracker Redis keys for this instance.

        Called during model deletion to clean up tracking data.

        Args:
            pipeline: Optional Redis pipeline for batch operations.
        """
        keys = [
            self._at_key("staged"),
            self._at_key("access_log"),
            self._at_key("meta"),
        ]
        if pipeline:
            for key in keys:
                pipeline.delete(key)
        else:
            get_REDIS_DB().delete(*keys)

access_count property

Total number of confirmed read accesses.

Returns:

Name	Type	Description
int		The cumulative access count, or 0 if never confirmed.

last_accessed property

Timestamp of the most recent confirmed read access.

Returns:

Type	Description
	float or None: Unix timestamp, or None if never confirmed.

export_state(model_instance, **kwargs) classmethod

Export the access-tracker meta counters and confirmed log.

Returns:

Type	Description
	``{"access_count": int, "last_accessed": float,
	"access_log": [float, ...]}, orNone`` when this instance has
	never had a confirmed access. Each key is omitted when its source
	structure is absent, so an instance with counters but an empty log
	exports exactly the shape #554-era code produced.

Source code in src/popoto/fields/access_tracker.py

@classmethod
def export_state(cls, model_instance, **kwargs):
    """Export the access-tracker meta counters and confirmed log.

    Returns:
        ``{"access_count": int, "last_accessed": float,
        "access_log": [float, ...]}``, or ``None`` when this instance has
        never had a confirmed access. Each key is omitted when its source
        structure is absent, so an instance with counters but an empty log
        exports exactly the shape #554-era code produced.
    """
    meta_key = model_instance._at_key("meta")
    raw_count = get_REDIS_DB().hget(meta_key, "access_count")
    raw_last = get_REDIS_DB().hget(meta_key, "last_accessed")
    raw_log = get_REDIS_DB().lrange(model_instance._at_key("access_log"), 0, -1)
    if raw_count is None and raw_last is None and not raw_log:
        return None

    state: dict[str, object] = {}
    if raw_count is not None:
        try:
            state["access_count"] = int(raw_count)
        except (TypeError, ValueError):
            logger.warning(f"Non-numeric access_count in {meta_key}; skipping")
    if raw_last is not None:
        try:
            state["last_accessed"] = float(raw_last)
        except (TypeError, ValueError):
            logger.warning(f"Non-numeric last_accessed in {meta_key}; skipping")
    if raw_log:
        # Decoded to floats here rather than left as bytes: to_jsonable
        # tags bytes with a __bytes__ envelope, which round-trips but
        # stops matching the plain-list shape the docs describe.
        log: list[float] = []
        for raw_ts in raw_log:
            try:
                log.append(float(raw_ts))
            except (TypeError, ValueError):
                logger.warning(
                    f"Non-numeric access_log entry for {meta_key}; skipping entry"
                )
        if log:
            state["access_log"] = log
    return state or None

import_state(model_instance, state, **kwargs) classmethod

Restore the access-tracker meta counters and confirmed log.

access_count and last_accessed are exposed only as read-only properties, so this writes the meta hash directly -- the same hash CONFIRM_ACCESS_LUA maintains. The log is restored by a raw DELETE + RPUSH rather than by staging the timestamps and calling confirm_access(): that path would HINCRBY the count a second time on top of the carried value and overwrite last_accessed, and it would push the timestamps through a staging list whose TTL contract has nothing to do with a transfer.

The log is trimmed to the destination's _max_access_log, keeping the most recent entries -- the same trim CONFIRM_ACCESS_LUA applies at :47. access_log absent from state is tolerated: a file exported by #554-era code has no such key, and must import cleanly rather than raise.

Source code in src/popoto/fields/access_tracker.py

@classmethod
def import_state(cls, model_instance, state, **kwargs):
    """Restore the access-tracker meta counters and confirmed log.

    ``access_count`` and ``last_accessed`` are exposed only as read-only
    properties, so this writes the meta hash directly -- the same hash
    ``CONFIRM_ACCESS_LUA`` maintains. The log is restored by a raw
    DELETE + RPUSH rather than by staging the timestamps and calling
    ``confirm_access()``: that path would HINCRBY the count a second time
    on top of the carried value and overwrite ``last_accessed``, and it
    would push the timestamps through a staging list whose TTL contract
    has nothing to do with a transfer.

    The log is trimmed to the *destination's* ``_max_access_log``, keeping
    the most recent entries -- the same trim ``CONFIRM_ACCESS_LUA``
    applies at ``:47``. ``access_log`` absent from ``state`` is tolerated:
    a file exported by #554-era code has no such key, and must import
    cleanly rather than raise.
    """
    if not state:
        return None

    mapping: dict[str, Any] = {}
    if state.get("access_count") is not None:
        mapping["access_count"] = int(state["access_count"])
    if state.get("last_accessed") is not None:
        mapping["last_accessed"] = float(state["last_accessed"])
    if mapping:
        get_REDIS_DB().hset(model_instance._at_key("meta"), mapping=mapping)

    log = state.get("access_log")
    if isinstance(log, (list, tuple)) and log:
        cap = max(1, int(cls._max_access_log))
        entries = [str(float(ts)) for ts in log][-cap:]
        log_key = model_instance._at_key("access_log")
        get_REDIS_DB().delete(log_key)
        get_REDIS_DB().rpush(log_key, *entries)
    return None

on_read(pipeline=None)

Record a read access by staging a timestamp.

Appends the current timestamp to the staging list and refreshes the TTL on the staged key. This is a cheap operation suitable for fire-and-forget use from query hooks.

Staged-read TTL contract

Applications MUST call confirm_access() within _staged_ttl_seconds (default 24h) of staging reads via on_read(). Reads left unconfirmed past the TTL window are permanently dropped from access_count / last_accessed by design. Set _staged_ttl_seconds higher if a longer stage→confirm cadence is required. The _staged_ttl_seconds constant is a magic-number tuning knob — it is not user-facing configuration.

Parameters:

Name	Type	Description	Default
pipeline		Optional Redis pipeline for batch operations. When provided, the RPUSH and EXPIRE are queued in the same pipeline call so they execute atomically.	None

Source code in src/popoto/fields/access_tracker.py

def on_read(self, pipeline=None):
    """Record a read access by staging a timestamp.

    Appends the current timestamp to the staging list and refreshes the
    TTL on the staged key. This is a cheap operation suitable for
    fire-and-forget use from query hooks.

    Staged-read TTL contract:
        Applications MUST call ``confirm_access()`` within
        ``_staged_ttl_seconds`` (default 24h) of staging reads via
        ``on_read()``. Reads left unconfirmed past the TTL window are
        permanently dropped from ``access_count`` / ``last_accessed`` by
        design. Set ``_staged_ttl_seconds`` higher if a longer
        stage→confirm cadence is required. The ``_staged_ttl_seconds``
        constant is a magic-number tuning knob — it is not user-facing
        configuration.

    Args:
        pipeline: Optional Redis pipeline for batch operations. When
            provided, the RPUSH and EXPIRE are queued in the same pipeline
            call so they execute atomically.
    """
    backend = non_redis_backend(self)
    if backend is not None:
        # #759 M2a: _staged_reads/_staged_at on the record's own row.
        self._access_call(
            backend,
            "stage",
            {self._at_member(): 1},
            now=time.time(),
            ttl=self._staged_ttl_seconds,
            uow=_uow_of(pipeline, backend),
        )
        return
    ts = str(time.time())
    staged_key = self._at_key("staged")
    if pipeline:
        pipeline.rpush(staged_key, ts)
        pipeline.expire(staged_key, self._staged_ttl_seconds)
    else:
        get_REDIS_DB().rpush(staged_key, ts)
        get_REDIS_DB().expire(staged_key, self._staged_ttl_seconds)

confirm_access(pipeline=None)

Atomically promote staged reads to the confirmed access log.

Uses a Lua script to: 1. Move all staged timestamps to the access log 2. Trim the log to max_access_log entries 3. Update access_count and last_accessed in the meta hash 4. Delete the staging list

TTL contract (confirm side): A staged key that expired before confirm contributes 0 to access_count and does not update last_accessed — this is by design (see on_read() staged-read TTL contract). When this happens a DEBUG log is emitted.

Parameters:

Name	Type	Description	Default
pipeline		Optional Redis pipeline (not used for Lua eval, reserved for future use).	None

Returns:

Name	Type	Description
int		Number of staged reads that were promoted (0 if the staged key was empty or had already expired).

Raises:

Type	Description
TypeError	If the model instance has not been saved to Redis.

Source code in src/popoto/fields/access_tracker.py

def confirm_access(self, pipeline=None):
    """Atomically promote staged reads to the confirmed access log.

    Uses a Lua script to:
    1. Move all staged timestamps to the access log
    2. Trim the log to max_access_log entries
    3. Update access_count and last_accessed in the meta hash
    4. Delete the staging list

    TTL contract (confirm side):
        A staged key that expired before confirm contributes 0 to
        ``access_count`` and does not update ``last_accessed`` — this is
        by design (see on_read() staged-read TTL contract). When this
        happens a DEBUG log is emitted.

    Args:
        pipeline: Optional Redis pipeline (not used for Lua eval,
            reserved for future use).

    Returns:
        int: Number of staged reads that were promoted (0 if the staged
            key was empty or had already expired).

    Raises:
        TypeError: If the model instance has not been saved to Redis.
    """
    if not hasattr(self, "_redis_key") and not hasattr(self, "db_key"):
        raise TypeError("confirm_access() requires a saved model instance")
    try:
        redis_key = self._redis_key or self.db_key.redis_key
    except Exception:
        raise TypeError("confirm_access() requires a saved model instance")

    backend = non_redis_backend(self)
    if backend is not None:
        # #759 M2a: one UPDATE promotes the live staged reads.
        promoted = self._access_call(
            backend,
            "confirm",
            self._at_record_id(),
            now=time.time(),
            ttl=self._staged_ttl_seconds,
            uow=_uow_of(pipeline, backend),
        )
        if promoted is None:
            raise TypeError("confirm_access() requires a saved model instance")
        if promoted == 0:
            logger.debug(
                "AccessTracker: staged key empty/expired at confirm for %s — "
                "read dropped (TTL contract)",
                self._at_member(),
            )
        return promoted

    # Check if the model has been saved (has a valid redis key in the DB)
    if not get_REDIS_DB().exists(redis_key):
        raise TypeError("confirm_access() requires a saved model instance")

    staged_key = self._at_key("staged")
    log_key = self._at_key("access_log")
    meta_key = self._at_key("meta")

    count = run_lua(
        get_REDIS_DB(),
        CONFIRM_ACCESS_LUA,
        3,  # number of KEYS
        staged_key,
        log_key,
        meta_key,
        str(self._max_access_log),
    )
    count = int(count)
    if count == 0:
        logger.debug(
            "AccessTracker: staged key empty/expired at confirm for %s — read dropped (TTL contract)",
            self.db_key,
        )
    return count

discard_staged_access(pipeline=None)

Discard all staged reads without affecting confirmed data.

Parameters:

Name	Type	Description	Default
pipeline		Optional Redis pipeline for batch operations.	None

Source code in src/popoto/fields/access_tracker.py

def discard_staged_access(self, pipeline=None):
    """Discard all staged reads without affecting confirmed data.

    Args:
        pipeline: Optional Redis pipeline for batch operations.
    """
    backend = non_redis_backend(self)
    if backend is not None:
        self._access_call(
            backend, "discard", self._at_record_id(), uow=_uow_of(pipeline, backend)
        )
        return
    staged_key = self._at_key("staged")
    if pipeline:
        pipeline.delete(staged_key)
    else:
        get_REDIS_DB().delete(staged_key)