AlchemicalEntry

Bases: Generic[T]

An entry in a queue.

Attributes:

Name	Type	Description
entry_id	int	the identifier of the entry. Guaranteed unique per AlchemicalQueues instance.
enqueued_at	datetime	when the entry was added to the queue.
schedule_at	datetime | None	do not remove the entry from the queue before this time.
priority	int	the priority of the entry.
data	T	the data stored in this entry.
claim_token	int | None	identifies this specific claim, if this entry came from AlchemicalTaskQueue.get (None for an entry from plain AlchemicalQueue, or from put(), neither of which claim anything). Pass it to release()/discard()/extend() to prove you still hold this particular claim and not a since-expired one.

Source code in src/alchemical_queues/main.py

class AlchemicalEntry(Generic[T]):
    """An entry in a queue.

    Attributes:
        entry_id (int): the identifier of the entry. Guaranteed unique per [AlchemicalQueues][alchemical_queues.AlchemicalQueues] instance.
        enqueued_at (datetime): when the entry was added to the queue.
        schedule_at (datetime | None): do not remove the entry from the queue before this time.
        priority (int): the priority of the entry.
        data (T): the data stored in this entry.
        claim_token (int | None): identifies this specific claim, if this entry
            came from [AlchemicalTaskQueue.get][alchemical_queues.AlchemicalTaskQueue.get]
            (`None` for an entry from plain `AlchemicalQueue`, or from `put()`,
            neither of which claim anything). Pass it to
            `release()`/`discard()`/`extend()` to prove you still hold this
            particular claim and not a since-expired one.
    """

    __slots__ = (
        "data",
        "entry_id",
        "enqueued_at",
        "schedule_at",
        "priority",
        "claim_token",
    )

    def __init__(
        self,
        entry,
        data: T,
    ):
        assert isinstance(entry.entry_id, int)

        self.entry_id: int = entry.entry_id
        self.enqueued_at: datetime = entry.enqueued_at
        self.schedule_at: Union[datetime, None] = entry.schedule_at
        self.claim_token: Union[int, None] = entry.claim_token
        self.priority: int = entry.priority
        self.data: T = data

    def __repr__(self):
        return (
            f"<{self.__class__.__module__}.{self.__class__.__name__} "
            f"entry_id={self.entry_id} enqueued_at={self.enqueued_at} "
            f"schedule_at={self.schedule_at} priority={self.priority}>"
        )