AsyncAlchemicalQueues

The async entrypoint to Alchemical Queues, built on an AsyncEngine. See AlchemicalQueues for the sync equivalent -- the two have the same API, methods here are just coroutines.

Source code in src/alchemical_queues/aio.py

class AsyncAlchemicalQueues:
    """The async entrypoint to Alchemical Queues, built on an `AsyncEngine`.
    See [AlchemicalQueues][alchemical_queues.AlchemicalQueues] for the sync
    equivalent -- the two have the same API, methods here are just
    coroutines.
    """

    def __init__(
        self,
        engine: Union[AsyncEngine, None] = None,
        queue_tablename: str = "AlchemicalQueue",
        response_tablename: str = "AlchemicalResult",
        base: Union[Type[DeclarativeBase], None] = None,
    ) -> None:
        """Create the async queue entrypoint object.

        Args:
            engine (AsyncEngine | None): The SQLAlchemy async engine you want
                to use (e.g. `create_async_engine("sqlite+aiosqlite:///...")`).
                May be left None and initialized later with `set_engine`.
            queue_tablename (str): see [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
            response_tablename (str): see [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
            base (Type[DeclarativeBase] | None, optional): see
                [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
        """
        self._engine = engine
        self._base, self._qmodel, self._rmodel = _generate_models(
            queue_tablename, response_tablename, base
        )
        self._queues: Dict[str, "AsyncAlchemicalQueue"] = {}
        self._task_queues: Dict[str, "AsyncAlchemicalTaskQueue"] = {}

    def set_engine(self, engine: AsyncEngine) -> None:
        """Set the SQLAlchemy async engine post-initialization.

        Args:
            engine (AsyncEngine): The SQLAlchemy async engine you want to use.

        Raises:
            Exception: when the engine was already set.
        """
        if self._engine is not None:
            raise Exception(
                "Cannot set the engine on AsyncAlchemicalQueues more than once!"
            )

        self._engine = engine

    async def create_all(self) -> None:
        """Create the needed SQLAlchemy tables. You would normally await this
        when you are also creating your own tables."""
        if self._engine is None:
            raise Exception(
                "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
            )

        async with self._engine.begin() as conn:
            await conn.run_sync(self._base.metadata.create_all)

    @_retry_on_deadlock
    async def clear(self) -> None:
        """Clear all entries from all queues and task results. Might fail-silent an update call."""
        if self._engine is None:
            raise Exception(
                "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
            )

        async with AsyncSession(self._engine) as session:
            await session.execute(delete(self._qmodel))
            await session.execute(delete(self._rmodel))
            await session.commit()

    def get(
        self, key: str, *, serializer: Serializer[Any] = DEFAULT_SERIALIZER
    ) -> "AsyncAlchemicalQueue[Any]":
        """Get a plain async Queue instance. See
        [AlchemicalQueues.get][alchemical_queues.AlchemicalQueues.get] for the
        sync equivalent.

        Returns:
            AsyncAlchemicalQueue
        """
        if self._engine is None:
            raise Exception(
                "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
            )

        if key not in self._queues:
            _validate_queue_name(key)
            self._queues[key] = AsyncAlchemicalQueue(
                self._engine, self._qmodel, key, serializer=serializer
            )

        return self._queues[key]

    def get_typed(
        self,
        key: str,
        typeof: Type[T],
        *,
        serializer: Union[Serializer[T], None] = None,
    ) -> "AsyncAlchemicalQueue[T]":
        """Get a typed async Queue instance. See
        [AlchemicalQueues.get_typed][alchemical_queues.AlchemicalQueues.get_typed]
        for the sync equivalent.

        Returns:
            AsyncAlchemicalQueue[T]
        """
        # pylint: disable=unused-argument
        return cast(
            "AsyncAlchemicalQueue[T]",
            self.get(key, serializer=serializer or DEFAULT_SERIALIZER),
        )

    def get_serialized(
        self, key: str, serializer: Serializer[T]
    ) -> "AsyncAlchemicalQueue[T]":
        """Get an async Queue instance typed by its `serializer`. See
        [AlchemicalQueues.get_serialized][alchemical_queues.AlchemicalQueues.get_serialized]
        for the sync equivalent.

        Returns:
            AsyncAlchemicalQueue[T]
        """
        return cast("AsyncAlchemicalQueue[T]", self.get(key, serializer=serializer))

    def get_task_queue(
        self,
        key: str,
        *,
        visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
        serializer: Serializer[Any] = DEFAULT_SERIALIZER,
        response_serializer: Serializer[Any] = DEFAULT_SERIALIZER,
    ) -> "AsyncAlchemicalTaskQueue[Any, Any]":
        """Get an async task queue instance. See
        [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]
        for the sync equivalent.

        Returns:
            AsyncAlchemicalTaskQueue
        """
        if self._engine is None:
            raise Exception(
                "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
            )

        if key not in self._task_queues:
            _validate_queue_name(key)
            self._task_queues[key] = AsyncAlchemicalTaskQueue(
                self._engine,
                self._qmodel,
                self._rmodel,
                key,
                visibility_timeout,
                serializer=serializer,
                response_serializer=response_serializer,
            )

        return self._task_queues[key]

    def get_task_queue_typed(
        self,
        key: str,
        typeof: Type[T],
        *,
        visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
        serializer: Union[Serializer[T], None] = None,
    ) -> "AsyncAlchemicalTaskQueue[T, Any]":
        """Get a typed async task queue instance. See
        [AlchemicalQueues.get_task_queue_typed][alchemical_queues.AlchemicalQueues.get_task_queue_typed]
        for the sync equivalent.

        Returns:
            AsyncAlchemicalTaskQueue[T, Any]
        """
        # pylint: disable=unused-argument
        return cast(
            "AsyncAlchemicalTaskQueue[T, Any]",
            self.get_task_queue(
                key,
                visibility_timeout=visibility_timeout,
                serializer=serializer or DEFAULT_SERIALIZER,
            ),
        )

    def get_task_queue_serialized(
        self,
        key: str,
        serializer: Serializer[T],
        *,
        response_serializer: Union[Serializer[R], None] = None,
        visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
    ) -> "AsyncAlchemicalTaskQueue[T, R]":
        """Get an async task queue instance typed by its `serializer`/
        `response_serializer`. See
        [AlchemicalQueues.get_task_queue_serialized][alchemical_queues.AlchemicalQueues.get_task_queue_serialized]
        for the sync equivalent.

        Returns:
            AsyncAlchemicalTaskQueue[T, R]
        """
        return cast(
            "AsyncAlchemicalTaskQueue[T, R]",
            self.get_task_queue(
                key,
                visibility_timeout=visibility_timeout,
                serializer=serializer,
                response_serializer=response_serializer or DEFAULT_SERIALIZER,
            ),
        )

__init__(engine=None, queue_tablename='AlchemicalQueue', response_tablename='AlchemicalResult', base=None)

Create the async queue entrypoint object.

Parameters:

Name	Type	Description	Default
engine	AsyncEngine | None	The SQLAlchemy async engine you want to use (e.g. create_async_engine("sqlite+aiosqlite:///...")). May be left None and initialized later with set_engine.	None
queue_tablename	str	see AlchemicalQueues.	'AlchemicalQueue'
response_tablename	str	see AlchemicalQueues.	'AlchemicalResult'
base	Type[DeclarativeBase] | None	see AlchemicalQueues.	None

Source code in src/alchemical_queues/aio.py

def __init__(
    self,
    engine: Union[AsyncEngine, None] = None,
    queue_tablename: str = "AlchemicalQueue",
    response_tablename: str = "AlchemicalResult",
    base: Union[Type[DeclarativeBase], None] = None,
) -> None:
    """Create the async queue entrypoint object.

    Args:
        engine (AsyncEngine | None): The SQLAlchemy async engine you want
            to use (e.g. `create_async_engine("sqlite+aiosqlite:///...")`).
            May be left None and initialized later with `set_engine`.
        queue_tablename (str): see [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
        response_tablename (str): see [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
        base (Type[DeclarativeBase] | None, optional): see
            [AlchemicalQueues][alchemical_queues.AlchemicalQueues].
    """
    self._engine = engine
    self._base, self._qmodel, self._rmodel = _generate_models(
        queue_tablename, response_tablename, base
    )
    self._queues: Dict[str, "AsyncAlchemicalQueue"] = {}
    self._task_queues: Dict[str, "AsyncAlchemicalTaskQueue"] = {}

set_engine(engine)

Set the SQLAlchemy async engine post-initialization.

Parameters:

Name	Type	Description	Default
engine	AsyncEngine	The SQLAlchemy async engine you want to use.	required

Raises:

Type	Description
Exception	when the engine was already set.

Source code in src/alchemical_queues/aio.py

def set_engine(self, engine: AsyncEngine) -> None:
    """Set the SQLAlchemy async engine post-initialization.

    Args:
        engine (AsyncEngine): The SQLAlchemy async engine you want to use.

    Raises:
        Exception: when the engine was already set.
    """
    if self._engine is not None:
        raise Exception(
            "Cannot set the engine on AsyncAlchemicalQueues more than once!"
        )

    self._engine = engine

create_all() async

Create the needed SQLAlchemy tables. You would normally await this when you are also creating your own tables.

Source code in src/alchemical_queues/aio.py

async def create_all(self) -> None:
    """Create the needed SQLAlchemy tables. You would normally await this
    when you are also creating your own tables."""
    if self._engine is None:
        raise Exception(
            "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
        )

    async with self._engine.begin() as conn:
        await conn.run_sync(self._base.metadata.create_all)

clear() async

Clear all entries from all queues and task results. 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 all queues and task results. Might fail-silent an update call."""
    if self._engine is None:
        raise Exception(
            "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
        )

    async with AsyncSession(self._engine) as session:
        await session.execute(delete(self._qmodel))
        await session.execute(delete(self._rmodel))
        await session.commit()

get(key, *, serializer=DEFAULT_SERIALIZER)

Get a plain async Queue instance. See AlchemicalQueues.get for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalQueue[Any]	AsyncAlchemicalQueue

Source code in src/alchemical_queues/aio.py

def get(
    self, key: str, *, serializer: Serializer[Any] = DEFAULT_SERIALIZER
) -> "AsyncAlchemicalQueue[Any]":
    """Get a plain async Queue instance. See
    [AlchemicalQueues.get][alchemical_queues.AlchemicalQueues.get] for the
    sync equivalent.

    Returns:
        AsyncAlchemicalQueue
    """
    if self._engine is None:
        raise Exception(
            "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
        )

    if key not in self._queues:
        _validate_queue_name(key)
        self._queues[key] = AsyncAlchemicalQueue(
            self._engine, self._qmodel, key, serializer=serializer
        )

    return self._queues[key]

get_typed(key, typeof, *, serializer=None)

Get a typed async Queue instance. See AlchemicalQueues.get_typed for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalQueue[T]	AsyncAlchemicalQueue[T]

Source code in src/alchemical_queues/aio.py

def get_typed(
    self,
    key: str,
    typeof: Type[T],
    *,
    serializer: Union[Serializer[T], None] = None,
) -> "AsyncAlchemicalQueue[T]":
    """Get a typed async Queue instance. See
    [AlchemicalQueues.get_typed][alchemical_queues.AlchemicalQueues.get_typed]
    for the sync equivalent.

    Returns:
        AsyncAlchemicalQueue[T]
    """
    # pylint: disable=unused-argument
    return cast(
        "AsyncAlchemicalQueue[T]",
        self.get(key, serializer=serializer or DEFAULT_SERIALIZER),
    )

get_serialized(key, serializer)

Get an async Queue instance typed by its serializer. See AlchemicalQueues.get_serialized for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalQueue[T]	AsyncAlchemicalQueue[T]

Source code in src/alchemical_queues/aio.py

def get_serialized(
    self, key: str, serializer: Serializer[T]
) -> "AsyncAlchemicalQueue[T]":
    """Get an async Queue instance typed by its `serializer`. See
    [AlchemicalQueues.get_serialized][alchemical_queues.AlchemicalQueues.get_serialized]
    for the sync equivalent.

    Returns:
        AsyncAlchemicalQueue[T]
    """
    return cast("AsyncAlchemicalQueue[T]", self.get(key, serializer=serializer))

get_task_queue(key, *, visibility_timeout=DEFAULT_VISIBILITY_TIMEOUT, serializer=DEFAULT_SERIALIZER, response_serializer=DEFAULT_SERIALIZER)

Get an async task queue instance. See AlchemicalQueues.get_task_queue for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalTaskQueue[Any, Any]	AsyncAlchemicalTaskQueue

Source code in src/alchemical_queues/aio.py

def get_task_queue(
    self,
    key: str,
    *,
    visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
    serializer: Serializer[Any] = DEFAULT_SERIALIZER,
    response_serializer: Serializer[Any] = DEFAULT_SERIALIZER,
) -> "AsyncAlchemicalTaskQueue[Any, Any]":
    """Get an async task queue instance. See
    [AlchemicalQueues.get_task_queue][alchemical_queues.AlchemicalQueues.get_task_queue]
    for the sync equivalent.

    Returns:
        AsyncAlchemicalTaskQueue
    """
    if self._engine is None:
        raise Exception(
            "AsyncAlchemicalQueues SQLAlchemy engine was not initialized."
        )

    if key not in self._task_queues:
        _validate_queue_name(key)
        self._task_queues[key] = AsyncAlchemicalTaskQueue(
            self._engine,
            self._qmodel,
            self._rmodel,
            key,
            visibility_timeout,
            serializer=serializer,
            response_serializer=response_serializer,
        )

    return self._task_queues[key]

get_task_queue_typed(key, typeof, *, visibility_timeout=DEFAULT_VISIBILITY_TIMEOUT, serializer=None)

Get a typed async task queue instance. See AlchemicalQueues.get_task_queue_typed for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalTaskQueue[T, Any]	AsyncAlchemicalTaskQueue[T, Any]

Source code in src/alchemical_queues/aio.py

def get_task_queue_typed(
    self,
    key: str,
    typeof: Type[T],
    *,
    visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
    serializer: Union[Serializer[T], None] = None,
) -> "AsyncAlchemicalTaskQueue[T, Any]":
    """Get a typed async task queue instance. See
    [AlchemicalQueues.get_task_queue_typed][alchemical_queues.AlchemicalQueues.get_task_queue_typed]
    for the sync equivalent.

    Returns:
        AsyncAlchemicalTaskQueue[T, Any]
    """
    # pylint: disable=unused-argument
    return cast(
        "AsyncAlchemicalTaskQueue[T, Any]",
        self.get_task_queue(
            key,
            visibility_timeout=visibility_timeout,
            serializer=serializer or DEFAULT_SERIALIZER,
        ),
    )

get_task_queue_serialized(key, serializer, *, response_serializer=None, visibility_timeout=DEFAULT_VISIBILITY_TIMEOUT)

Get an async task queue instance typed by its serializer/ response_serializer. See AlchemicalQueues.get_task_queue_serialized for the sync equivalent.

Returns:

Type	Description
AsyncAlchemicalTaskQueue[T, R]	AsyncAlchemicalTaskQueue[T, R]

Source code in src/alchemical_queues/aio.py

def get_task_queue_serialized(
    self,
    key: str,
    serializer: Serializer[T],
    *,
    response_serializer: Union[Serializer[R], None] = None,
    visibility_timeout: timedelta = DEFAULT_VISIBILITY_TIMEOUT,
) -> "AsyncAlchemicalTaskQueue[T, R]":
    """Get an async task queue instance typed by its `serializer`/
    `response_serializer`. See
    [AlchemicalQueues.get_task_queue_serialized][alchemical_queues.AlchemicalQueues.get_task_queue_serialized]
    for the sync equivalent.

    Returns:
        AsyncAlchemicalTaskQueue[T, R]
    """
    return cast(
        "AsyncAlchemicalTaskQueue[T, R]",
        self.get_task_queue(
            key,
            visibility_timeout=visibility_timeout,
            serializer=serializer,
            response_serializer=response_serializer or DEFAULT_SERIALIZER,
        ),
    )