AsyncWorker

Async equivalent of tasks.Worker.

Attributes:

Name	Type	Description
queue	AsyncAlchemicalTaskQueue	the queue this worker runs on
poll_every	timedelta	how often to poll for new tasks
keepalive_every	timedelta | None	how often to extend a task's claim while it's still running. See __init__.

Source code in src/alchemical_queues/tasks/aio.py

class AsyncWorker:
    """Async equivalent of [tasks.Worker][alchemical_queues.tasks.Worker].

    Attributes:
        queue (AsyncAlchemicalTaskQueue): the queue this worker runs on
        poll_every (timedelta): how often to poll for new tasks
        keepalive_every (timedelta | None): how often to extend a task's claim
            while it's still running. See `__init__`.
    """

    def __init__(
        self,
        queue: AsyncAlchemicalTaskQueue,
        poll_every: timedelta = timedelta(seconds=1),
        *,
        keepalive_every: Union[timedelta, None] = None,
    ):
        """
        Args:
            queue (AsyncAlchemicalTaskQueue): the queue this worker runs on.
                Obtain one via
                [AsyncAlchemicalQueues.get_task_queue][alchemical_queues.aio.AsyncAlchemicalQueues.get_task_queue].
            poll_every (timedelta, optional): how often to poll for new tasks
                when the queue is empty.
            keepalive_every (timedelta | None, optional): if set, a background
                asyncio task extends a task's claim by this often while its
                handler is still running -- see
                [tasks.Worker][alchemical_queues.tasks.Worker] for why you'd
                want this. Left `None` (the default), a task running longer
                than `visibility_timeout` risks its result being discarded
                rather than responded with, to avoid risking a duplicate.

        Only `async def` handlers registered via `async_task` can be run by
        this worker -- a plain sync handler registered with
        [task][alchemical_queues.tasks.task] needs the sync
        [Worker][alchemical_queues.tasks.Worker] instead.
        """
        self.queue = queue
        self.poll_every: timedelta = poll_every
        self.keepalive_every = keepalive_every
        self._handler_registry: Dict[str, "AsyncTasker"] = {}
        self._logger = getLogger("alchemical_queues.tasks")

    async def _fail(
        self,
        entry_id: int,
        data: Dict[str, Any],
        exception: Union[BaseException, None] = None,
        fatal: bool = False,
    ):
        retries = data["retries"]

        if data.get("max_retries", 0) > retries and not fatal:
            data["retries"] = retries + 1

            retry_at = datetime.now()
            if data["retry_in"] is not None:
                retry_at += data["retry_in"]

            data["entry_id"] = entry_id
            new_entry = await self.queue.put(data, schedule_at=retry_at)
            self._logger.info(
                "Retrying failed task %s as `%s`", entry_id, new_entry.entry_id
            )
            return False

        self._logger.warning("Failed to perform task %s", entry_id)
        self._logger.exception(exception)
        await self.queue.respond(
            entry_id,
            {
                "error": str(exception),
                "error_type": type(exception).__name__ if exception else None,
            },
        )

        return False

    async def _keepalive_loop(self, task_entry: AlchemicalEntry) -> None:
        assert self.keepalive_every is not None
        assert task_entry.claim_token is not None
        while True:
            await asyncio.sleep(self.keepalive_every.total_seconds())
            try:
                await self.queue.extend(task_entry.entry_id, task_entry.claim_token)
            except ClaimExpired:
                # Nothing more we can do here -- the main task's own
                # discard() call will discover the same thing and log it
                # once, so don't duplicate the warning.
                return

    async def _perform(self, task_entry: AlchemicalEntry):
        # task_entry always comes from self.queue.get(), which always sets a
        # claim_token -- unlike a bare put() result, where it would be None.
        assert task_entry.claim_token is not None
        claim_token: int = task_entry.claim_token

        data = task_entry.data
        entry_id = task_entry.data.get("entry_id") or task_entry.entry_id
        function_path = data["function"]

        task_handler = self._handler_registry.get(function_path)
        if function_path not in self._handler_registry:
            task_handler = self._handler_registry[function_path] = cast(
                AsyncTasker, locate(function_path)
            )

        succeeded = False
        result: Any = None
        error: Union[BaseException, None] = None
        fatal = False

        if task_handler is None:
            error = KeyError(f"AlchemicalEntry handler `{function_path}` not found.")
            fatal = True
        else:
            keepalive_task: Union["asyncio.Task[None]", None] = None
            if self.keepalive_every is not None:
                keepalive_task = asyncio.create_task(self._keepalive_loop(task_entry))

            try:
                self._logger.info("Running task `%s`.", task_entry.entry_id)
                func = task_handler.get_handler()
                result = await func(
                    TaskInfo(task_entry.entry_id, data["retries"], data["max_retries"]),
                    *data["args"],
                    **data["kwargs"],
                )
                succeeded = True
            except asyncio.CancelledError:
                # Allow cancellation (e.g. worker shutdown), but let the
                # claim go back to the queue immediately rather than leaving
                # it to time out, since we're not actually going to finish it.
                try:
                    await self.queue.release(task_entry.entry_id, claim_token)
                except ClaimExpired:
                    pass
                raise
            except Exception as caught:  # pylint: disable=broad-except
                error = caught
            finally:
                if keepalive_task is not None:
                    keepalive_task.cancel()
                    try:
                        await keepalive_task
                    except asyncio.CancelledError:
                        pass

        # Fencing gate: only act on the outcome above if we still hold this
        # exact claim -- see Worker._perform (tasks/main.py) for why.
        try:
            await self.queue.discard(task_entry.entry_id, claim_token)
        except ClaimExpired:
            self._logger.warning(
                "Claim for task %s expired before we could act on its outcome "
                "(it may already have been redelivered and handled elsewhere); "
                "discarding our own result rather than risking a duplicate. If "
                "this task can run longer than the queue's visibility_timeout, "
                "pass keepalive_every= to AsyncWorker() or raise visibility_timeout.",
                task_entry.entry_id,
            )
            return False

        if succeeded:
            await self.queue.respond(entry_id, {"result": result})
            return True

        return await self._fail(entry_id, data, error, fatal=fatal)

    async def work(self) -> NoReturn:
        """Run tasks forever."""
        self._logger.info("AsyncWorker starting on queue `%s`.", self.queue.name)

        while True:
            task_entry = await self.queue.get()

            if task_entry is None:
                await asyncio.sleep(self.poll_every.total_seconds())
            else:
                await self._perform(task_entry)

    async def work_one(self, block: bool = True) -> None:
        """Run exactly one task.

        Args:
            block (bool): whether to wait until a task is available, or
                return immediately if none is available.
        """
        while True:
            task_entry = await self.queue.get()

            if task_entry is not None:
                await self._perform(task_entry)
                return

            if block:
                await asyncio.sleep(self.poll_every.total_seconds())
            else:
                break

__init__(queue, poll_every=timedelta(seconds=1), *, keepalive_every=None)

Parameters:

Name	Type	Description	Default
queue	AsyncAlchemicalTaskQueue	the queue this worker runs on. Obtain one via AsyncAlchemicalQueues.get_task_queue.	required
poll_every	timedelta	how often to poll for new tasks when the queue is empty.	timedelta(seconds=1)
keepalive_every	timedelta | None	if set, a background asyncio task extends a task's claim by this often while its handler is still running -- see tasks.Worker for why you'd want this. Left None (the default), a task running longer than visibility_timeout risks its result being discarded rather than responded with, to avoid risking a duplicate.	None

Only async def handlers registered via async_task can be run by this worker -- a plain sync handler registered with task needs the sync Worker instead.

Source code in src/alchemical_queues/tasks/aio.py

def __init__(
    self,
    queue: AsyncAlchemicalTaskQueue,
    poll_every: timedelta = timedelta(seconds=1),
    *,
    keepalive_every: Union[timedelta, None] = None,
):
    """
    Args:
        queue (AsyncAlchemicalTaskQueue): the queue this worker runs on.
            Obtain one via
            [AsyncAlchemicalQueues.get_task_queue][alchemical_queues.aio.AsyncAlchemicalQueues.get_task_queue].
        poll_every (timedelta, optional): how often to poll for new tasks
            when the queue is empty.
        keepalive_every (timedelta | None, optional): if set, a background
            asyncio task extends a task's claim by this often while its
            handler is still running -- see
            [tasks.Worker][alchemical_queues.tasks.Worker] for why you'd
            want this. Left `None` (the default), a task running longer
            than `visibility_timeout` risks its result being discarded
            rather than responded with, to avoid risking a duplicate.

    Only `async def` handlers registered via `async_task` can be run by
    this worker -- a plain sync handler registered with
    [task][alchemical_queues.tasks.task] needs the sync
    [Worker][alchemical_queues.tasks.Worker] instead.
    """
    self.queue = queue
    self.poll_every: timedelta = poll_every
    self.keepalive_every = keepalive_every
    self._handler_registry: Dict[str, "AsyncTasker"] = {}
    self._logger = getLogger("alchemical_queues.tasks")

work() async

Run tasks forever.

Source code in src/alchemical_queues/tasks/aio.py

async def work(self) -> NoReturn:
    """Run tasks forever."""
    self._logger.info("AsyncWorker starting on queue `%s`.", self.queue.name)

    while True:
        task_entry = await self.queue.get()

        if task_entry is None:
            await asyncio.sleep(self.poll_every.total_seconds())
        else:
            await self._perform(task_entry)

work_one(block=True) async

Run exactly one task.

Parameters:

Name	Type	Description	Default
block	bool	whether to wait until a task is available, or return immediately if none is available.	True

Source code in src/alchemical_queues/tasks/aio.py

async def work_one(self, block: bool = True) -> None:
    """Run exactly one task.

    Args:
        block (bool): whether to wait until a task is available, or
            return immediately if none is available.
    """
    while True:
        task_entry = await self.queue.get()

        if task_entry is not None:
            await self._perform(task_entry)
            return

        if block:
            await asyncio.sleep(self.poll_every.total_seconds())
        else:
            break