Getting started¶

Install¶

pip install asynk

or, with uv:

uv add asynk

Your first app¶

An App ties together a transport (where messages travel) and a store (where results are kept). This one uses the in-memory transport and store, so everything runs in a single process.

"""The smallest asynk app: one task, published and run in the same process."""

import asyncio

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

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


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


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    handle = await add.delay(2, b=3)
    print("2 + 3 =", await handle.result(timeout=5))

    worker.request_shutdown()
    await run


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

@app.task() registers a function as a task. add.delay(2, b=3) publishes a message and returns a handle. A Worker consumes the message and runs the task.

Defining tasks¶

A task is an async def function; registering a plain def raises TypeError. Default argument values work as you would expect, and retry retries a failing task, here up to three times (see Retries).

"""Tasks are async functions and can have default arguments."""

import asyncio
from typing import Any

from asynk import App, MemoryStore, RetryPolicy, TaskHandle, TaskOptions, Worker
from asynk.transport.memory import memory_transport

producer, consumer = memory_transport()
app = App("defining-tasks", producer=producer, consumer=consumer, store=MemoryStore())


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


@app.task()
async def shout(text: str) -> str:
    return text.upper()


@app.task(options=TaskOptions(retry=RetryPolicy(max_retries=3)))
async def greet(name: str, greeting: str = "Hello") -> str:
    return f"{greeting}, {name}!"


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    handles: list[TaskHandle[Any]] = [
        await add.delay(1, 2),
        await shout.delay("hello"),
        await greet.delay("Ada"),
    ]
    for handle in handles:
        print(await handle.result(timeout=5))

    worker.request_shutdown()
    await run


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

Blocking code¶

A blocking call inside an async task blocks the worker's event loop, and every other task with it, unless you offload it:

"""A blocking call in an async task is offloaded to a thread."""

import asyncio
import time

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

producer, consumer = memory_transport()
app = App("blocking-task", producer=producer, consumer=consumer, store=MemoryStore())


@app.task()
async def slow_square(n: int) -> int:
    # time.sleep(0.1) here would block the event loop; to_thread does not.
    await asyncio.to_thread(time.sleep, 0.1)
    return n * n


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    handle = await slow_square.delay(4)
    print(await handle.result(timeout=5))

    worker.request_shutdown()
    await run


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

A thread cannot be cancelled. On shutdown or timeout the task is cancelled, but the thread keeps running until the call returns.

Calling tasks¶

delay takes the same arguments as the function and is type-checked against its signature. It returns a handle you can use to wait for the result or read the task state.

"""Calling a task with delay and reading its result from the store."""

import asyncio

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

producer, consumer = memory_transport()
app = App("call-tasks", producer=producer, consumer=consumer, store=MemoryStore())


@app.task()
async def greet(name: str, greeting: str = "Hello") -> str:
    return f"{greeting}, {name}!"


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    # Positional and keyword arguments, type-checked against the task signature.
    first = await greet.delay("Ada")
    second = await greet.delay("Grace", greeting="Welcome")

    # Each call returns a handle that reads the result from the app's store.
    print(await first.result(timeout=5))
    print(await second.result(timeout=5))
    print((await second.state()).value)

    worker.request_shutdown()
    await run


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

Running a worker¶

With a transport that connects separate processes, you run workers from the command line. Point the CLI at the module and the app inside it:

asynk worker tasks:app --concurrency 10

What next¶

Concepts explains the envelope, codec, transport and store. The guides cover scheduled tasks, results and the store and deployment. See the API reference for every public class and function.

The repository's examples/memory/ runs a worker and a scheduler in one process, with a fast virtual clock.