AlchemicalTaskQueue

Bases: Generic[T, R]

An Alchemical Queue with crash-safe delivery: get() claims an entry rather than removing it, release()/discard()/extend() manage that claim, and respond()/responses() record and retrieve a result for a given entry. This is the queue tasks.Worker is built on. It is not intended to be initialized by a user, go through AlchemicalQueues.get_task_queue instead.

T is the type of entries (put()/get()); R is the type of responses (respond()/responses()) -- they're independent, each with their own serializer/response_serializer, since a task's input and its result are usually different shapes.

If you just want a plain FIFO/priority queue and don't need redelivery, retries, or result tracking, use AlchemicalQueues.get / AlchemicalQueue instead -- it has a much smaller surface.

Source code in src/alchemical_queues/main.py

class AlchemicalTaskQueue(Generic[T, R]):
    """An Alchemical Queue with crash-safe delivery: `get()` claims an entry
    rather than removing it, `release()`/`discard()`/`extend()` manage that
    claim, and `respond()`/`responses()` record and retrieve a result for a
    given entry. This is the queue [tasks.Worker][alchemical_queues.tasks.Worker]
    is built on. It is not intended to be initialized by a user, go through
    [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]
    instead.

    `T` is the type of entries (`put()`/`get()`); `R` is the type of
    responses (`respond()`/`responses()`) -- they're independent, each with
    their own `serializer`/`response_serializer`, since a task's input and
    its result are usually different shapes.

    If you just want a plain FIFO/priority queue and don't need redelivery,
    retries, or result tracking, use
    [AlchemicalQueues.get][alchemical_queues.AlchemicalQueues.get] /
    [AlchemicalQueue][alchemical_queues.AlchemicalQueue] instead -- it has a
    much smaller surface.
    """

    def __init__(
        self,
        engine: Engine,
        model,
        response_model,
        name: str,
        visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
        serializer: Serializer[T] = DEFAULT_SERIALIZER,
        response_serializer: Serializer[R] = DEFAULT_SERIALIZER,
    ):
        self._engine = engine
        self._model = model
        self._response_model = response_model
        self._name = name
        self._visibility_timeout = visibility_timeout
        self._response_serializer = response_serializer
        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

    @property
    def visibility_timeout(self) -> timedelta:
        """The default visibility_timeout this queue's `get()` calls use."""
        return self._visibility_timeout

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

        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 claimed.
            priority (int, optional): Entry priority. Entries are claimed 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, *, visibility_timeout: Union[timedelta, None] = None
    ) -> Union["AlchemicalEntry[T]", None]:
        """Claim the highest priority entry from the queue.

        The entry is not removed by this call: it is marked claimed until
        `visibility_timeout` elapses, so if your process dies before you call
        [release][alchemical_queues.AlchemicalTaskQueue.release] or
        [discard][alchemical_queues.AlchemicalTaskQueue.discard] for it, a
        later `get()` call will eventually claim it again instead of the
        entry being lost. Call `discard()` (after `respond()`, if you have a
        result) or `release()` (if you want it retried sooner than the
        timeout) as soon as you're done with an entry to free it up
        immediately, rather than waiting out the timeout. For work that can
        run longer than `visibility_timeout`, call
        [extend][alchemical_queues.AlchemicalTaskQueue.extend] periodically to
        push the deadline out while you're still on it.

        The returned entry's `claim_token` identifies *this* claim
        specifically -- pass it to `release()`/`discard()`/`extend()` so they
        only affect it, even if your claim has since timed out and been
        reclaimed by someone else (same entry_id, different claim_token).

        Args:
            visibility_timeout (timedelta | None, optional): how long this entry
                stays claimed before it becomes claimable again. Defaults to the
                visibility_timeout this queue was obtained with (see
                [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]).

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

        timestamp = datetime.now()
        timeout = (
            visibility_timeout
            if visibility_timeout is not None
            else self._visibility_timeout
        )
        token = _new_claim_token()

        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
                    ),
                    or_(
                        self._model.claimed_until == None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                        self._model.claimed_until <= 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, "update"):
                # Picking the candidate row and claiming it happen in one
                # atomic UPDATE ... RETURNING statement, so two concurrent
                # get() calls can never claim the same entry. This needs no
                # special transaction isolation (previously SQLite needed a
                # global BEGIN EXCLUSIVE, which serialized every transaction
                # on the engine, including ones from unrelated code sharing
                # the same engine).
                candidate = query.with_only_columns(self._model.entry_id)
                item = session.execute(
                    update(self._model)
                    .where(self._model.entry_id == candidate.scalar_subquery())
                    .values(claimed_until=timestamp + timeout, claim_token=token)
                    .returning(self._model)
                ).scalar_one_or_none()
            else:
                # MySQL has no RETURNING support at all, and MariaDB only
                # supports it for DELETE/INSERT, not UPDATE: lock and fetch
                # the full candidate row first instead, set the claim fields
                # on the already-loaded instance, and let the session flush
                # that as a plain UPDATE on commit. The row stays locked
                # throughout, so the same atomicity guarantee holds.
                item = session.execute(query).scalar_one_or_none()
                if item is not None:
                    item.claimed_until = timestamp + timeout
                    item.claim_token = token

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

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

        return entry

    @_retry_on_deadlock
    def release(self, entry_id: int, claim_token: int) -> None:
        """Release a claimed entry back to the queue, without recording a
        response for it. The entry itself is kept -- it becomes claimable
        again immediately, same as if its visibility timeout had just
        elapsed, instead of whoever's turn it is next having to wait that
        out.

        Use this when you called [get][alchemical_queues.AlchemicalTaskQueue.get],
        decided you have nothing to respond with yet (you want it retried
        sooner than the timeout, or by someone else), and don't want to
        discard it. To remove an entry entirely without a response, see
        [discard][alchemical_queues.AlchemicalTaskQueue.discard].

        Args:
            entry_id (int): the entry_id you wish to release.
            claim_token (int): the claim_token from the `get()` call that
                claimed this entry (`AlchemicalEntry.claim_token`).

        Raises:
            ClaimExpired: if entry_id isn't currently claimed under claim_token.
        """

        if not isinstance(entry_id, int):
            raise TypeError(f"entry_id={entry_id} should be integer")

        with self._session() as session:
            result = session.execute(
                update(self._model)
                .where(
                    self._model.entry_id == entry_id,
                    self._model.claim_token == claim_token,
                )
                .values(claimed_until=None, claim_token=None)
            )
            session.commit()

        if result.rowcount == 0:  # type: ignore[attr-defined]
            raise ClaimExpired(entry_id)

    @_retry_on_deadlock
    def discard(self, entry_id: int, claim_token: int) -> None:
        """Remove a claimed entry from the queue entirely, without recording
        a response for it. Unlike [release][alchemical_queues.AlchemicalTaskQueue.release],
        the entry does not become claimable again -- it's simply gone, the same
        as if [respond][alchemical_queues.AlchemicalTaskQueue.respond] had been
        called but without creating a response.

        Args:
            entry_id (int): the entry_id you wish to discard.
            claim_token (int): the claim_token from the `get()` call that
                claimed this entry (`AlchemicalEntry.claim_token`).

        Raises:
            ClaimExpired: if entry_id isn't currently claimed under claim_token.
        """

        if not isinstance(entry_id, int):
            raise TypeError(f"entry_id={entry_id} should be integer")

        with self._session() as session:
            result = session.execute(
                delete(self._model).where(
                    self._model.entry_id == entry_id,
                    self._model.claim_token == claim_token,
                )
            )
            session.commit()

        if result.rowcount == 0:  # type: ignore[attr-defined]
            raise ClaimExpired(entry_id)

    @_retry_on_deadlock
    def extend(
        self,
        entry_id: int,
        claim_token: int,
        *,
        by: Union[timedelta, None] = None,
    ) -> None:
        """Extend a claim's visibility timeout, for work that can take longer
        than it. Call this periodically while you're still actively on an
        entry (a keepalive) to push its claim's expiry further out, so it
        isn't redelivered to another `get()` call while you're still working
        on it. [tasks.Worker][alchemical_queues.tasks.Worker] can do this for
        you automatically; see its `keepalive_every` argument.

        Args:
            entry_id (int): the entry_id whose claim you wish to extend.
            claim_token (int): the claim_token from the `get()` call that
                claimed this entry (`AlchemicalEntry.claim_token`).
            by (timedelta | None, optional): how much longer the claim should
                last from now. Defaults to this queue's configured visibility_timeout.

        Raises:
            ClaimExpired: if entry_id isn't currently claimed under claim_token
                -- it's already timed out (and possibly been redelivered and
                finished by someone else), so extending it further would be
                meaningless at best and misleading at worst. Stop working.
        """

        if not isinstance(entry_id, int):
            raise TypeError(f"entry_id={entry_id} should be integer")

        timeout = by if by is not None else self._visibility_timeout
        new_claimed_until = datetime.now() + timeout

        with self._session() as session:
            result = session.execute(
                update(self._model)
                .where(
                    self._model.entry_id == entry_id,
                    self._model.claim_token == claim_token,
                )
                .values(claimed_until=new_claimed_until)
            )
            session.commit()

        if result.rowcount == 0:  # type: ignore[attr-defined]
            raise ClaimExpired(entry_id)

    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()

    @_retry_on_deadlock
    def respond(
        self, entry_id: int, response: R, cleanup_at: Union[datetime, None] = None
    ) -> "AlchemicalResponse[R]":
        """Record a response for a queue entry. Used to implement task queues.

        This only records the response -- it has no effect on a claim you may
        be holding on entry_id (there may not even be one: it's independent
        of [get][alchemical_queues.AlchemicalTaskQueue.get]/claim lifecycle
        entirely, just like `responses()` is). Call
        [discard][alchemical_queues.AlchemicalTaskQueue.discard] separately once
        you've responded, to free up the entry you claimed.

        If this entry_id belongs to a `tasks.task`-scheduled task, note that
        [QueuedTask.result][alchemical_queues.tasks.QueuedTask.result] only
        recognizes the `{"result": ...}`/`{"error": ..., "error_type": ...}`
        shape `tasks.Worker` itself responds with to tell success from
        failure -- call `respond()` directly (as here) with anything else
        and `QueuedTask.result` just hands that value back to you unparsed.

        Args:
            entry_id (int): The entry_id you wish to respond to.
            response (R): The response data. Must be serializable by this
                queue's `response_serializer` (pickle-able, by default).
            cleanup_at (datetime, optional): The optional cleanup timestamp. After this time the response will be removed.
                                             By default it is not automatically cleaned up.

        Returns:
            AlchemicalResponse[R]: the response as sent.
        """

        if not isinstance(entry_id, int):
            raise TypeError(f"entry_id={entry_id} should be integer")

        entry = self._response_model(
            entry_id=entry_id,
            delivered_at=datetime.now(),
            cleanup_at=cleanup_at,
            queue_name=self._name,
            data=self._response_serializer.dumps(response),
        )

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

            return AlchemicalResponse(entry, response)

    @_retry_on_deadlock
    def responses(self, entry_id: int) -> List["AlchemicalResponse[R]"]:
        """Obtain the response(s) to a specific queue entry.

        Returns:
            List[AlchemicalResponse[R]]: A list of responses
        """
        if not isinstance(entry_id, int):
            raise TypeError(f"entry_id={entry_id} should be integer")

        with self._session() as session:
            now = datetime.now()

            session.execute(
                delete(self._response_model).where(
                    self._response_model.cleanup_at != None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                    self._response_model.cleanup_at < now,
                )
            )
            session.commit()

            entries: Any = session.scalars(
                select(self._response_model).where(
                    self._response_model.queue_name == self._name,
                    self._response_model.entry_id == entry_id,
                )
            ).all()
            return [
                AlchemicalResponse(e, self._response_serializer.loads(e.data))
                for e in entries
            ]

name property

The name of the queue

visibility_timeout property

The default visibility_timeout this queue's get() calls use.

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

Put an entry into the queue

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 claimed.	None
priority	int	Entry priority. Entries are claimed 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 queue

    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 claimed.
        priority (int, optional): Entry priority. Entries are claimed 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(*, visibility_timeout=None)

Claim the highest priority entry from the queue.

The entry is not removed by this call: it is marked claimed until visibility_timeout elapses, so if your process dies before you call release or discard for it, a later get() call will eventually claim it again instead of the entry being lost. Call discard() (after respond(), if you have a result) or release() (if you want it retried sooner than the timeout) as soon as you're done with an entry to free it up immediately, rather than waiting out the timeout. For work that can run longer than visibility_timeout, call extend periodically to push the deadline out while you're still on it.

The returned entry's claim_token identifies this claim specifically -- pass it to release()/discard()/extend() so they only affect it, even if your claim has since timed out and been reclaimed by someone else (same entry_id, different claim_token).

Parameters:

Name	Type	Description	Default
visibility_timeout	timedelta | None	how long this entry stays claimed before it becomes claimable again. Defaults to the visibility_timeout this queue was obtained with (see AlchemicalQueues.get_task_queue).	None

Returns:

Type	Description
AlchemicalEntry | None	The claimed entry, or None if the queue is empty
Union[AlchemicalEntry[T], None]	(or nothing is claimable/scheduled yet)

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def get(
    self, *, visibility_timeout: Union[timedelta, None] = None
) -> Union["AlchemicalEntry[T]", None]:
    """Claim the highest priority entry from the queue.

    The entry is not removed by this call: it is marked claimed until
    `visibility_timeout` elapses, so if your process dies before you call
    [release][alchemical_queues.AlchemicalTaskQueue.release] or
    [discard][alchemical_queues.AlchemicalTaskQueue.discard] for it, a
    later `get()` call will eventually claim it again instead of the
    entry being lost. Call `discard()` (after `respond()`, if you have a
    result) or `release()` (if you want it retried sooner than the
    timeout) as soon as you're done with an entry to free it up
    immediately, rather than waiting out the timeout. For work that can
    run longer than `visibility_timeout`, call
    [extend][alchemical_queues.AlchemicalTaskQueue.extend] periodically to
    push the deadline out while you're still on it.

    The returned entry's `claim_token` identifies *this* claim
    specifically -- pass it to `release()`/`discard()`/`extend()` so they
    only affect it, even if your claim has since timed out and been
    reclaimed by someone else (same entry_id, different claim_token).

    Args:
        visibility_timeout (timedelta | None, optional): how long this entry
            stays claimed before it becomes claimable again. Defaults to the
            visibility_timeout this queue was obtained with (see
            [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]).

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

    timestamp = datetime.now()
    timeout = (
        visibility_timeout
        if visibility_timeout is not None
        else self._visibility_timeout
    )
    token = _new_claim_token()

    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
                ),
                or_(
                    self._model.claimed_until == None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                    self._model.claimed_until <= 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, "update"):
            # Picking the candidate row and claiming it happen in one
            # atomic UPDATE ... RETURNING statement, so two concurrent
            # get() calls can never claim the same entry. This needs no
            # special transaction isolation (previously SQLite needed a
            # global BEGIN EXCLUSIVE, which serialized every transaction
            # on the engine, including ones from unrelated code sharing
            # the same engine).
            candidate = query.with_only_columns(self._model.entry_id)
            item = session.execute(
                update(self._model)
                .where(self._model.entry_id == candidate.scalar_subquery())
                .values(claimed_until=timestamp + timeout, claim_token=token)
                .returning(self._model)
            ).scalar_one_or_none()
        else:
            # MySQL has no RETURNING support at all, and MariaDB only
            # supports it for DELETE/INSERT, not UPDATE: lock and fetch
            # the full candidate row first instead, set the claim fields
            # on the already-loaded instance, and let the session flush
            # that as a plain UPDATE on commit. The row stays locked
            # throughout, so the same atomicity guarantee holds.
            item = session.execute(query).scalar_one_or_none()
            if item is not None:
                item.claimed_until = timestamp + timeout
                item.claim_token = token

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

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

    return entry

release(entry_id, claim_token)

Release a claimed entry back to the queue, without recording a response for it. The entry itself is kept -- it becomes claimable again immediately, same as if its visibility timeout had just elapsed, instead of whoever's turn it is next having to wait that out.

Use this when you called get, decided you have nothing to respond with yet (you want it retried sooner than the timeout, or by someone else), and don't want to discard it. To remove an entry entirely without a response, see discard.

Parameters:

Name	Type	Description	Default
entry_id	int	the entry_id you wish to release.	required
claim_token	int	the claim_token from the get() call that claimed this entry (AlchemicalEntry.claim_token).	required

Raises:

Type	Description
ClaimExpired	if entry_id isn't currently claimed under claim_token.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def release(self, entry_id: int, claim_token: int) -> None:
    """Release a claimed entry back to the queue, without recording a
    response for it. The entry itself is kept -- it becomes claimable
    again immediately, same as if its visibility timeout had just
    elapsed, instead of whoever's turn it is next having to wait that
    out.

    Use this when you called [get][alchemical_queues.AlchemicalTaskQueue.get],
    decided you have nothing to respond with yet (you want it retried
    sooner than the timeout, or by someone else), and don't want to
    discard it. To remove an entry entirely without a response, see
    [discard][alchemical_queues.AlchemicalTaskQueue.discard].

    Args:
        entry_id (int): the entry_id you wish to release.
        claim_token (int): the claim_token from the `get()` call that
            claimed this entry (`AlchemicalEntry.claim_token`).

    Raises:
        ClaimExpired: if entry_id isn't currently claimed under claim_token.
    """

    if not isinstance(entry_id, int):
        raise TypeError(f"entry_id={entry_id} should be integer")

    with self._session() as session:
        result = session.execute(
            update(self._model)
            .where(
                self._model.entry_id == entry_id,
                self._model.claim_token == claim_token,
            )
            .values(claimed_until=None, claim_token=None)
        )
        session.commit()

    if result.rowcount == 0:  # type: ignore[attr-defined]
        raise ClaimExpired(entry_id)

discard(entry_id, claim_token)

Remove a claimed entry from the queue entirely, without recording a response for it. Unlike release, the entry does not become claimable again -- it's simply gone, the same as if respond had been called but without creating a response.

Parameters:

Name	Type	Description	Default
entry_id	int	the entry_id you wish to discard.	required
claim_token	int	the claim_token from the get() call that claimed this entry (AlchemicalEntry.claim_token).	required

Raises:

Type	Description
ClaimExpired	if entry_id isn't currently claimed under claim_token.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def discard(self, entry_id: int, claim_token: int) -> None:
    """Remove a claimed entry from the queue entirely, without recording
    a response for it. Unlike [release][alchemical_queues.AlchemicalTaskQueue.release],
    the entry does not become claimable again -- it's simply gone, the same
    as if [respond][alchemical_queues.AlchemicalTaskQueue.respond] had been
    called but without creating a response.

    Args:
        entry_id (int): the entry_id you wish to discard.
        claim_token (int): the claim_token from the `get()` call that
            claimed this entry (`AlchemicalEntry.claim_token`).

    Raises:
        ClaimExpired: if entry_id isn't currently claimed under claim_token.
    """

    if not isinstance(entry_id, int):
        raise TypeError(f"entry_id={entry_id} should be integer")

    with self._session() as session:
        result = session.execute(
            delete(self._model).where(
                self._model.entry_id == entry_id,
                self._model.claim_token == claim_token,
            )
        )
        session.commit()

    if result.rowcount == 0:  # type: ignore[attr-defined]
        raise ClaimExpired(entry_id)

extend(entry_id, claim_token, *, by=None)

Extend a claim's visibility timeout, for work that can take longer than it. Call this periodically while you're still actively on an entry (a keepalive) to push its claim's expiry further out, so it isn't redelivered to another get() call while you're still working on it. tasks.Worker can do this for you automatically; see its keepalive_every argument.

Parameters:

Name	Type	Description	Default
entry_id	int	the entry_id whose claim you wish to extend.	required
claim_token	int	the claim_token from the get() call that claimed this entry (AlchemicalEntry.claim_token).	required
by	timedelta | None	how much longer the claim should last from now. Defaults to this queue's configured visibility_timeout.	None

Raises:

Type	Description
ClaimExpired	if entry_id isn't currently claimed under claim_token -- it's already timed out (and possibly been redelivered and finished by someone else), so extending it further would be meaningless at best and misleading at worst. Stop working.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def extend(
    self,
    entry_id: int,
    claim_token: int,
    *,
    by: Union[timedelta, None] = None,
) -> None:
    """Extend a claim's visibility timeout, for work that can take longer
    than it. Call this periodically while you're still actively on an
    entry (a keepalive) to push its claim's expiry further out, so it
    isn't redelivered to another `get()` call while you're still working
    on it. [tasks.Worker][alchemical_queues.tasks.Worker] can do this for
    you automatically; see its `keepalive_every` argument.

    Args:
        entry_id (int): the entry_id whose claim you wish to extend.
        claim_token (int): the claim_token from the `get()` call that
            claimed this entry (`AlchemicalEntry.claim_token`).
        by (timedelta | None, optional): how much longer the claim should
            last from now. Defaults to this queue's configured visibility_timeout.

    Raises:
        ClaimExpired: if entry_id isn't currently claimed under claim_token
            -- it's already timed out (and possibly been redelivered and
            finished by someone else), so extending it further would be
            meaningless at best and misleading at worst. Stop working.
    """

    if not isinstance(entry_id, int):
        raise TypeError(f"entry_id={entry_id} should be integer")

    timeout = by if by is not None else self._visibility_timeout
    new_claimed_until = datetime.now() + timeout

    with self._session() as session:
        result = session.execute(
            update(self._model)
            .where(
                self._model.entry_id == entry_id,
                self._model.claim_token == claim_token,
            )
            .values(claimed_until=new_claimed_until)
        )
        session.commit()

    if result.rowcount == 0:  # type: ignore[attr-defined]
        raise ClaimExpired(entry_id)

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()

respond(entry_id, response, cleanup_at=None)

Record a response for a queue entry. Used to implement task queues.

This only records the response -- it has no effect on a claim you may be holding on entry_id (there may not even be one: it's independent of get/claim lifecycle entirely, just like responses() is). Call discard separately once you've responded, to free up the entry you claimed.

If this entry_id belongs to a tasks.task-scheduled task, note that QueuedTask.result only recognizes the {"result": ...}/{"error": ..., "error_type": ...} shape tasks.Worker itself responds with to tell success from failure -- call respond() directly (as here) with anything else and QueuedTask.result just hands that value back to you unparsed.

Parameters:

Name	Type	Description	Default
entry_id	int	The entry_id you wish to respond to.	required
response	R	The response data. Must be serializable by this queue's response_serializer (pickle-able, by default).	required
cleanup_at	datetime	The optional cleanup timestamp. After this time the response will be removed. By default it is not automatically cleaned up.	None

Returns:

Type	Description
AlchemicalResponse[R]	AlchemicalResponse[R]: the response as sent.

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def respond(
    self, entry_id: int, response: R, cleanup_at: Union[datetime, None] = None
) -> "AlchemicalResponse[R]":
    """Record a response for a queue entry. Used to implement task queues.

    This only records the response -- it has no effect on a claim you may
    be holding on entry_id (there may not even be one: it's independent
    of [get][alchemical_queues.AlchemicalTaskQueue.get]/claim lifecycle
    entirely, just like `responses()` is). Call
    [discard][alchemical_queues.AlchemicalTaskQueue.discard] separately once
    you've responded, to free up the entry you claimed.

    If this entry_id belongs to a `tasks.task`-scheduled task, note that
    [QueuedTask.result][alchemical_queues.tasks.QueuedTask.result] only
    recognizes the `{"result": ...}`/`{"error": ..., "error_type": ...}`
    shape `tasks.Worker` itself responds with to tell success from
    failure -- call `respond()` directly (as here) with anything else
    and `QueuedTask.result` just hands that value back to you unparsed.

    Args:
        entry_id (int): The entry_id you wish to respond to.
        response (R): The response data. Must be serializable by this
            queue's `response_serializer` (pickle-able, by default).
        cleanup_at (datetime, optional): The optional cleanup timestamp. After this time the response will be removed.
                                         By default it is not automatically cleaned up.

    Returns:
        AlchemicalResponse[R]: the response as sent.
    """

    if not isinstance(entry_id, int):
        raise TypeError(f"entry_id={entry_id} should be integer")

    entry = self._response_model(
        entry_id=entry_id,
        delivered_at=datetime.now(),
        cleanup_at=cleanup_at,
        queue_name=self._name,
        data=self._response_serializer.dumps(response),
    )

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

        return AlchemicalResponse(entry, response)

responses(entry_id)

Obtain the response(s) to a specific queue entry.

Returns:

Type	Description
List[AlchemicalResponse[R]]	List[AlchemicalResponse[R]]: A list of responses

Source code in src/alchemical_queues/main.py

@_retry_on_deadlock
def responses(self, entry_id: int) -> List["AlchemicalResponse[R]"]:
    """Obtain the response(s) to a specific queue entry.

    Returns:
        List[AlchemicalResponse[R]]: A list of responses
    """
    if not isinstance(entry_id, int):
        raise TypeError(f"entry_id={entry_id} should be integer")

    with self._session() as session:
        now = datetime.now()

        session.execute(
            delete(self._response_model).where(
                self._response_model.cleanup_at != None,  # noqa: E711 (SQLAlchemy requires == None for IS NULL)
                self._response_model.cleanup_at < now,
            )
        )
        session.commit()

        entries: Any = session.scalars(
            select(self._response_model).where(
                self._response_model.queue_name == self._name,
                self._response_model.entry_id == entry_id,
            )
        ).all()
        return [
            AlchemicalResponse(e, self._response_serializer.loads(e.data))
            for e in entries
        ]