AsyncAlchemicalQueue

Bases: Generic[T]

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

Source code in src/alchemical_queues/aio.py

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

    def __init__(
        self,
        engine: AsyncEngine,
        model,
        name: str,
        serializer: Serializer[T] = DEFAULT_SERIALIZER,
    ):
        self._engine = engine
        self._model = model
        self._name = name
        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

    @_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
        [AlchemicalQueue.put][alchemical_queues.AlchemicalQueue.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) -> Union["AlchemicalEntry[T]", None]:
        """Get the highest priority entry out from the queue, removing it.
        See [AlchemicalQueue.get][alchemical_queues.AlchemicalQueue.get] for
        details.

        Returns:
            (AlchemicalEntry | None)
        """
        timestamp = datetime.now()

        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
                    ),
                )
                .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"):
                candidate = query.with_only_columns(self._model.entry_id)
                item = (
                    await 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 = (await session.execute(query)).scalar_one_or_none()
                if item is not None:
                    await session.delete(item)

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

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

        return entry

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

name property

The name of the queue

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

Put an entry into the queue. See AlchemicalQueue.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
    [AlchemicalQueue.put][alchemical_queues.AlchemicalQueue.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() async

Get the highest priority entry out from the queue, removing it. See AlchemicalQueue.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) -> Union["AlchemicalEntry[T]", None]:
    """Get the highest priority entry out from the queue, removing it.
    See [AlchemicalQueue.get][alchemical_queues.AlchemicalQueue.get] for
    details.

    Returns:
        (AlchemicalEntry | None)
    """
    timestamp = datetime.now()

    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
                ),
            )
            .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"):
            candidate = query.with_only_columns(self._model.entry_id)
            item = (
                await 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 = (await session.execute(query)).scalar_one_or_none()
            if item is not None:
                await session.delete(item)

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

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

    return entry

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