AlchemicalQueue

Bases: Generic[T]

A plain Alchemical Queue: put an item in, get the highest-priority one back out. get() removes the entry immediately -- there's no claim to manage, nothing to acknowledge, and no redelivery if your process dies partway through handling what it returned. It is not intended to be initialized by a user, go through AlchemicalQueues instead.

If you need crash-safe delivery (an entry comes back if the worker that claimed it dies before finishing), retries, or a way to record and poll for a result, use AlchemicalQueues.get_task_queue instead.

Source code in src/alchemical_queues/main.py

class AlchemicalQueue(Generic[T]):
    """A plain Alchemical Queue: put an item in, get the highest-priority one
    back out. `get()` removes the entry immediately -- there's no claim to
    manage, nothing to acknowledge, and no redelivery if your process dies
    partway through handling what it returned. It is not intended to be
    initialized by a user, go through
    [AlchemicalQueues][alchemical_queues.AlchemicalQueues] instead.

    If you need crash-safe delivery (an entry comes back if the worker that
    claimed it dies before finishing), retries, or a way to record and poll
    for a result, use
    [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]
    instead.
    """

    def __init__(
        self,
        engine: Engine,
        model,
        name: str,
        serializer: Serializer[T] = DEFAULT_SERIALIZER,
    ):
        self._engine = engine
        self._model = model
        self._name = name
        self._serializer = serializer
        self._session = sessionmaker(
            engine,
            autoflush=False,
            expire_on_commit=False,
        )

    @property
    def name(self) -> str:
        """The name of the queue"""
        return self._name

    @_retry_on_deadlock
    def put(
        self,
        item: T,
        *,
        schedule_at: Union[datetime, None] = None,
        priority: int = 0,
    ) -> "AlchemicalEntry[T]":
        """Put an entry into the AlchemicalQueue

        Args:
            item (T): The item you wish to add to the queue. Must be
                serializable by this queue's `serializer` (pickle-able, by
                default).
            schedule_at (datetime | None, optional): Earliest timestamp this entry may be popped of the queue.
            priority (int, optional): Entry priority. Entries are popped of first in order of priority and then
                                      in order of adding to the queue.

        Returns:
            AlchemicalEntry[T]: The resultant queue entry.
        """

        entry = self._model(
            enqueued_at=datetime.now(),
            schedule_at=schedule_at,
            priority=priority,
            queue_name=self._name,
            data=self._serializer.dumps(item),
        )

        with self._session() as session:
            session.add(entry)
            session.commit()

            return AlchemicalEntry(entry, item)

    @_retry_on_deadlock
    def get(self) -> Union["AlchemicalEntry[T]", None]:
        """Get the highest priority entry out from the queue, removing it.

        Returns:
            (AlchemicalEntry | None): The popped entry, or None if the queue is empty (or nothing is scheduled yet)
        """

        timestamp = datetime.now()

        with self._session() as session:
            query = (
                select(self._model)
                .filter(
                    self._model.queue_name == self._name,
                    or_(
                        self._model.schedule_at == None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                        self._model.schedule_at <= timestamp,  # type: ignore
                    ),
                )
                .order_by(self._model.priority.desc(), self._model.entry_id.asc())  # type: ignore
                .limit(1)
            )

            if self._engine.dialect.name in _SKIP_LOCKED_DIALECTS:
                query = query.with_for_update(skip_locked=True)

            if _supports_returning(self._engine, "delete"):
                # Picking the candidate row and deleting it happen in one
                # atomic DELETE ... RETURNING statement, so two concurrent
                # get() calls can never pop the same entry, with no extra
                # transaction isolation needed for that guarantee to hold.
                candidate = query.with_only_columns(self._model.entry_id)
                item = session.execute(
                    delete(self._model)
                    .where(self._model.entry_id == candidate.scalar_subquery())
                    .returning(self._model)
                ).scalar_one_or_none()
            else:
                # MySQL has no RETURNING support at all: lock and fetch the
                # full candidate row first instead, then delete it by id in
                # the same transaction. The row stays locked throughout, so
                # the same atomicity guarantee holds.
                item = session.execute(query).scalar_one_or_none()
                if item is not None:
                    session.delete(item)

            if item is None:
                session.rollback()
                return None

            entry = AlchemicalEntry(item, self._serializer.loads(item.data))
            session.commit()

        return entry

    def qsize(self) -> int:
        """Return the approximate size of this queue.

        Returns:
            int: Queue size.
        """
        with self._session() as session:
            return (
                session.scalar(
                    select(func.count())  # pylint: disable=not-callable
                    .select_from(self._model)
                    .where(self._model.queue_name == self._name)
                )
                or 0
            )

    def empty(self) -> bool:
        """Return `True` if the Queue is emtpy, `False` otherwise. More efficient than
        `qsize() > 0`.

        Returns:
            bool: wether the Queue is empty.
        """
        with self._session() as session:
            return (
                session.scalar(
                    select(self._model.entry_id)
                    .where(self._model.queue_name == self._name)
                    .limit(1)
                )
                is None
            )

    @_retry_on_deadlock
    def clear(self) -> None:
        """Clear all entries from this queue. Might fail-silent an update call."""

        with self._session() as session:
            session.execute(
                delete(self._model).where(self._model.queue_name == self._name)
            )
            session.commit()

name property

The name of the queue

put(item, *, schedule_at=None, priority=0)

Put an entry into the AlchemicalQueue

Parameters:

Name	Type	Description	Default
item	T	The item you wish to add to the queue. Must be serializable by this queue's serializer (pickle-able, by default).	required
schedule_at	datetime | None	Earliest timestamp this entry may be popped of the queue.	None
priority	int	Entry priority. Entries are popped of first in order of priority and then in order of adding to the queue.	0

Returns:

Type	Description
AlchemicalEntry[T]	AlchemicalEntry[T]: The resultant queue entry.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def put(
    self,
    item: T,
    *,
    schedule_at: Union[datetime, None] = None,
    priority: int = 0,
) -> "AlchemicalEntry[T]":
    """Put an entry into the AlchemicalQueue

    Args:
        item (T): The item you wish to add to the queue. Must be
            serializable by this queue's `serializer` (pickle-able, by
            default).
        schedule_at (datetime | None, optional): Earliest timestamp this entry may be popped of the queue.
        priority (int, optional): Entry priority. Entries are popped of first in order of priority and then
                                  in order of adding to the queue.

    Returns:
        AlchemicalEntry[T]: The resultant queue entry.
    """

    entry = self._model(
        enqueued_at=datetime.now(),
        schedule_at=schedule_at,
        priority=priority,
        queue_name=self._name,
        data=self._serializer.dumps(item),
    )

    with self._session() as session:
        session.add(entry)
        session.commit()

        return AlchemicalEntry(entry, item)

get()

Get the highest priority entry out from the queue, removing it.

Returns:

Type	Description
AlchemicalEntry | None	The popped entry, or None if the queue is empty (or nothing is scheduled yet)

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def get(self) -> Union["AlchemicalEntry[T]", None]:
    """Get the highest priority entry out from the queue, removing it.

    Returns:
        (AlchemicalEntry | None): The popped entry, or None if the queue is empty (or nothing is scheduled yet)
    """

    timestamp = datetime.now()

    with self._session() as session:
        query = (
            select(self._model)
            .filter(
                self._model.queue_name == self._name,
                or_(
                    self._model.schedule_at == None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                    self._model.schedule_at <= timestamp,  # type: ignore
                ),
            )
            .order_by(self._model.priority.desc(), self._model.entry_id.asc())  # type: ignore
            .limit(1)
        )

        if self._engine.dialect.name in _SKIP_LOCKED_DIALECTS:
            query = query.with_for_update(skip_locked=True)

        if _supports_returning(self._engine, "delete"):
            # Picking the candidate row and deleting it happen in one
            # atomic DELETE ... RETURNING statement, so two concurrent
            # get() calls can never pop the same entry, with no extra
            # transaction isolation needed for that guarantee to hold.
            candidate = query.with_only_columns(self._model.entry_id)
            item = session.execute(
                delete(self._model)
                .where(self._model.entry_id == candidate.scalar_subquery())
                .returning(self._model)
            ).scalar_one_or_none()
        else:
            # MySQL has no RETURNING support at all: lock and fetch the
            # full candidate row first instead, then delete it by id in
            # the same transaction. The row stays locked throughout, so
            # the same atomicity guarantee holds.
            item = session.execute(query).scalar_one_or_none()
            if item is not None:
                session.delete(item)

        if item is None:
            session.rollback()
            return None

        entry = AlchemicalEntry(item, self._serializer.loads(item.data))
        session.commit()

    return entry

qsize()

Return the approximate size of this queue.

Returns:

Name	Type	Description
int	int	Queue size.

Source code in src/alchemical_queues/main.py

def qsize(self) -> int:
    """Return the approximate size of this queue.

    Returns:
        int: Queue size.
    """
    with self._session() as session:
        return (
            session.scalar(
                select(func.count())  # pylint: disable=not-callable
                .select_from(self._model)
                .where(self._model.queue_name == self._name)
            )
            or 0
        )

empty()

Return True if the Queue is emtpy, False otherwise. More efficient than qsize() > 0.

Returns:

Name	Type	Description
bool	bool	wether the Queue is empty.

Source code in src/alchemical_queues/main.py

def empty(self) -> bool:
    """Return `True` if the Queue is emtpy, `False` otherwise. More efficient than
    `qsize() > 0`.

    Returns:
        bool: wether the Queue is empty.
    """
    with self._session() as session:
        return (
            session.scalar(
                select(self._model.entry_id)
                .where(self._model.queue_name == self._name)
                .limit(1)
            )
            is None
        )

clear()

Clear all entries from this queue. Might fail-silent an update call.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def clear(self) -> None:
    """Clear all entries from this queue. Might fail-silent an update call."""

    with self._session() as session:
        session.execute(
            delete(self._model).where(self._model.queue_name == self._name)
        )
        session.commit()