Results and the store¶

Pass a store to the app to record task states and return values. delay then returns a handle that reads them:

"""Reading task states and results from the app's store."""

import asyncio

from asynk import App, MemoryStore, TaskError, Worker
from asynk.transport.memory import memory_transport

producer, consumer = memory_transport()
app = App("results", producer=producer, consumer=consumer, store=MemoryStore())


@app.task()
async def add(a: int, b: int) -> int:
    return a + b


@app.task()
async def divide(a: int, b: int) -> float:
    return a / b


async def main() -> None:
    # Published before any worker runs, so the task is still pending.
    handle = await add.delay(1, b=2)
    print("before the worker:", (await handle.state()).value)

    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    print("result:", await handle.result(timeout=5))
    print("after the worker:", (await handle.state()).value)

    # Another request or process sharing the store only needs the id.
    same = app.get_task(handle.task_id)
    print("from the id:", await same.result(timeout=5))

    failing = await divide.delay(1, 0)
    try:
        await failing.result(timeout=5)
    except TaskError as exc:
        print("failed:", exc.type_name, "-", exc.message)
    print("state:", (await failing.state()).value)

    worker.request_shutdown()
    await run


if __name__ == "__main__":
    asyncio.run(main())

Handles¶

delay returns a TaskHandle[R], where R is the type the task returns, with:

• task_id, the envelope id;
• receipt, whatever the producer's publish returned;
• state(), the current TaskState;
• result(timeout=None), which waits for the task to finish and returns its value, or None if the task was skipped.

app.get_task(task_id) builds a handle from an id alone, for example in another request or process that shares the store. result(timeout=...) raises TimeoutError when the task is not done in time, and TaskNotFoundError for an id the store has no record of. A failed task makes result() raise TaskError, which carries the failure's type name and message as .type_name and .message.

Typed results¶

A task declared async def add(a: int, b: int) -> int makes add.delay(...) return a TaskHandle[int], so a type checker sees await handle.result() as int | None (None when a middleware skipped the task). app.get_task(id) and app.publish(...) know no task, so they return a TaskHandle[Any], which you can assign to a TaskHandle[int] or use untyped.

The type is a declaration. Asynk does not check or convert the value at runtime: result() returns whatever the codec decodes. Typed results are meant for values the codec carries as they are: None, booleans, numbers, strings, and lists and dicts of those. Anything else is not rebuilt as the declared type. With the default JsonCodec, a task declared to return a dataclass or a datetime produces a dict or a string, and a tuple comes back as a list, even though the checker believes result() is the declared type. Convert such a value yourself, for example Point(**result), or return plain values from the task. A task whose result the codec cannot encode fails with ResultNotSerializable (see below).

A task declared -> None has a TaskHandle[None], whose result() is always None. mypy reports using that value, as in assert await handle.result() is None, as "does not return a value"; annotate the handle as TaskHandle[Any] or just await it.

States¶

A failed attempt is final, FAILURE, when the task has no retry policy, when the error is not in the policy's retry_on, or when its retries are used up. Otherwise the state is RETRYING until the retry runs, and result() keeps waiting for a later attempt (see Retries). A message for a task name the worker does not know is recorded as FAILURE. A return value the codec cannot encode is recorded as a FAILURE with the type name ResultNotSerializable.

Store errors never affect a task or its delivery: the worker logs them and carries on, so a record can be missing after a store outage.

Store lifecycle¶

A worker and a scheduler start the app's store before their before_startup hooks and stop it after their after_shutdown hooks, also when startup fails, so they need nothing extra. A producer-only script, which runs neither, starts and stops the store itself (a fragment: it needs a worker running elsewhere):

await app.store.start()
handle = await add.delay(1, b=2)
print(await handle.result(timeout=10))
await app.store.stop()

Stores¶

MemoryStore(ttl=86400.0) keeps records in one process for ttl seconds, so the producer, the worker and readers must share that process. It is the development and testing store. For anything wider, use the Redis store, or implement the Store protocol for another backend; see Writing a custom store.

Without a store the app uses a NullStore: it records nothing, and state() and result() raise NotConfiguredError.

Codec¶

Task results and states are encoded with app.codec, the same codec that encodes messages (JsonCodec() by default). Pass App(..., codec=MyCodec()) to use another format, for example pickled values that keep tuples. With JsonCodec, a tuple comes back as a list.