AsyncAlchemicalTaskQueue

Bases: Generic[T, R]

Async equivalent of AlchemicalTaskQueue. Not intended to be initialized by a user, go through AsyncAlchemicalQueues.get_task_queue instead.

Source code in src/alchemical_queues/aio.py

class AsyncAlchemicalTaskQueue(Generic[T, R]):
    """Async equivalent of
    [AlchemicalTaskQueue][alchemical_queues.AlchemicalTaskQueue]. Not
    intended to be initialized by a user, go through
    [AsyncAlchemicalQueues.get_task_queue][alchemical_queues.aio.AsyncAlchemicalQueues.get_task_queue]
    instead.
    """

    def __init__(
        self,
        engine: AsyncEngine,
        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 = async_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
    async def put(
        self,
        item: T,
        *,
        schedule_at: Union[datetime, None] = None,
        priority: int = 0,
    ) -> "AlchemicalEntry[T]":
        """Put an entry into the queue. See
        [AlchemicalTaskQueue.put][alchemical_queues.AlchemicalTaskQueue.put]
        for details.

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

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

            return AlchemicalEntry(entry, item)

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

        Returns:
            (AlchemicalEntry | None)
        """
        timestamp = datetime.now()
        timeout = (
            visibility_timeout
            if visibility_timeout is not None
            else self._visibility_timeout
        )
        token = _new_claim_token()

        async with self._session() as session:
            query = (
                select(self._model)
                .filter(
                    self._model.queue_name == self._name,
                    or_(
                        self._model.schedule_at == None,  # noqa: E711
                        self._model.schedule_at <= timestamp,  # type: ignore
                    ),
                    or_(
                        self._model.claimed_until == None,  # noqa: E711
                        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"):
                candidate = query.with_only_columns(self._model.entry_id)
                item = (
                    await 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 = (await 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:
                await session.rollback()
                return None

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

        return entry

    @_retry_on_deadlock
    async def release(self, entry_id: int, claim_token: int) -> None:
        """Release a claimed entry back to the queue. See
        [AlchemicalTaskQueue.release][alchemical_queues.AlchemicalTaskQueue.release]
        for details.

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

        async with self._session() as session:
            result = await 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)
            )
            await session.commit()

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

    @_retry_on_deadlock
    async def discard(self, entry_id: int, claim_token: int) -> None:
        """Remove a claimed entry from the queue entirely. See
        [AlchemicalTaskQueue.discard][alchemical_queues.AlchemicalTaskQueue.discard]
        for details.

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

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

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

    @_retry_on_deadlock
    async def extend(
        self,
        entry_id: int,
        claim_token: int,
        *,
        by: Union[timedelta, None] = None,
    ) -> None:
        """Extend a claim's visibility timeout. See
        [AlchemicalTaskQueue.extend][alchemical_queues.AlchemicalTaskQueue.extend]
        for details.

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

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

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

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

    async def qsize(self) -> int:
        """Return the approximate size of this queue."""
        async with self._session() as session:
            return (
                await session.scalar(
                    select(func.count())  # pylint: disable=not-callable
                    .select_from(self._model)
                    .where(self._model.queue_name == self._name)
                )
                or 0
            )

    async def empty(self) -> bool:
        """Return `True` if the Queue is empty, `False` otherwise."""
        async with self._session() as session:
            return (
                await session.scalar(
                    select(self._model.entry_id)
                    .where(self._model.queue_name == self._name)
                    .limit(1)
                )
            ) is None

    @_retry_on_deadlock
    async def clear(self) -> None:
        """Clear all entries from this queue. Might fail-silent an update call."""
        async with self._session() as session:
            await session.execute(
                delete(self._model).where(self._model.queue_name == self._name)
            )
            await session.commit()

    @_retry_on_deadlock
    async def respond(
        self, entry_id: int, response: R, cleanup_at: Union[datetime, None] = None
    ) -> "AlchemicalResponse[R]":
        """Record a response for a queue entry. See
        [AlchemicalTaskQueue.respond][alchemical_queues.AlchemicalTaskQueue.respond]
        for details.

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

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

            return AlchemicalResponse(entry, response)

    @_retry_on_deadlock
    async 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")

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

            await session.execute(
                delete(self._response_model).where(
                    self._response_model.cleanup_at != None,  # noqa: E711
                    self._response_model.cleanup_at < now,
                )
            )
            await session.commit()

            entries: Any = (
                await 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) async

Put an entry into the queue. See AlchemicalTaskQueue.put for details.

Returns:

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

Source code in src/alchemical_queues/aio.py

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

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

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

        return AlchemicalEntry(entry, item)

get(*, visibility_timeout=None) async

Claim the highest priority entry from the queue. See AlchemicalTaskQueue.get for details.

Returns:

Type	Description
Union[AlchemicalEntry[T], None]	(AlchemicalEntry | None)

Source code in src/alchemical_queues/aio.py

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

    Returns:
        (AlchemicalEntry | None)
    """
    timestamp = datetime.now()
    timeout = (
        visibility_timeout
        if visibility_timeout is not None
        else self._visibility_timeout
    )
    token = _new_claim_token()

    async with self._session() as session:
        query = (
            select(self._model)
            .filter(
                self._model.queue_name == self._name,
                or_(
                    self._model.schedule_at == None,  # noqa: E711
                    self._model.schedule_at <= timestamp,  # type: ignore
                ),
                or_(
                    self._model.claimed_until == None,  # noqa: E711
                    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"):
            candidate = query.with_only_columns(self._model.entry_id)
            item = (
                await 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 = (await 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:
            await session.rollback()
            return None

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

    return entry

release(entry_id, claim_token) async

Release a claimed entry back to the queue. See AlchemicalTaskQueue.release for details.

Raises:

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

Source code in src/alchemical_queues/aio.py

@_retry_on_deadlock
async def release(self, entry_id: int, claim_token: int) -> None:
    """Release a claimed entry back to the queue. See
    [AlchemicalTaskQueue.release][alchemical_queues.AlchemicalTaskQueue.release]
    for details.

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

    async with self._session() as session:
        result = await 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)
        )
        await session.commit()

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

discard(entry_id, claim_token) async

Remove a claimed entry from the queue entirely. See AlchemicalTaskQueue.discard for details.

Raises:

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

Source code in src/alchemical_queues/aio.py

@_retry_on_deadlock
async def discard(self, entry_id: int, claim_token: int) -> None:
    """Remove a claimed entry from the queue entirely. See
    [AlchemicalTaskQueue.discard][alchemical_queues.AlchemicalTaskQueue.discard]
    for details.

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

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

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

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

Extend a claim's visibility timeout. See AlchemicalTaskQueue.extend for details.

Raises:

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

Source code in src/alchemical_queues/aio.py

@_retry_on_deadlock
async def extend(
    self,
    entry_id: int,
    claim_token: int,
    *,
    by: Union[timedelta, None] = None,
) -> None:
    """Extend a claim's visibility timeout. See
    [AlchemicalTaskQueue.extend][alchemical_queues.AlchemicalTaskQueue.extend]
    for details.

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

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

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

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

qsize() async

Return the approximate size of this queue.

Source code in src/alchemical_queues/aio.py

async def qsize(self) -> int:
    """Return the approximate size of this queue."""
    async with self._session() as session:
        return (
            await session.scalar(
                select(func.count())  # pylint: disable=not-callable
                .select_from(self._model)
                .where(self._model.queue_name == self._name)
            )
            or 0
        )

empty() async

Return True if the Queue is empty, False otherwise.

Source code in src/alchemical_queues/aio.py

async def empty(self) -> bool:
    """Return `True` if the Queue is empty, `False` otherwise."""
    async with self._session() as session:
        return (
            await session.scalar(
                select(self._model.entry_id)
                .where(self._model.queue_name == self._name)
                .limit(1)
            )
        ) is None

clear() async

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

Source code in src/alchemical_queues/aio.py

@_retry_on_deadlock
async def clear(self) -> None:
    """Clear all entries from this queue. Might fail-silent an update call."""
    async with self._session() as session:
        await session.execute(
            delete(self._model).where(self._model.queue_name == self._name)
        )
        await session.commit()

respond(entry_id, response, cleanup_at=None) async

Record a response for a queue entry. See AlchemicalTaskQueue.respond for details.

Returns:

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

Source code in src/alchemical_queues/aio.py

@_retry_on_deadlock
async def respond(
    self, entry_id: int, response: R, cleanup_at: Union[datetime, None] = None
) -> "AlchemicalResponse[R]":
    """Record a response for a queue entry. See
    [AlchemicalTaskQueue.respond][alchemical_queues.AlchemicalTaskQueue.respond]
    for details.

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

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

        return AlchemicalResponse(entry, response)

responses(entry_id) async

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/aio.py

@_retry_on_deadlock
async 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")

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

        await session.execute(
            delete(self._response_model).where(
                self._response_model.cleanup_at != None,  # noqa: E711
                self._response_model.cleanup_at < now,
            )
        )
        await session.commit()

        entries: Any = (
            await 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
        ]