Lifecycle hooks¶

Lifecycle hooks extend a worker or a scheduler without subclassing it, for example to open a connection pool at startup and close it at shutdown. To run code around each task instead, see Middleware.

"""Lifecycle hooks shared by every asynk process, branching on the runner's role."""

import asyncio

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

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

connections: list[str] = []


@app.before_startup
async def connect(runner: Runner) -> None:
    print(f"{runner.role}: setting up telemetry")
    if runner.role == "worker":
        connections.append("database")  # only workers need the database


@app.after_startup
def started(runner: Runner) -> None:
    print(f"{runner.role}: started")


@app.before_shutdown
def stopping(runner: Runner) -> None:
    print(f"{runner.role}: stopping")


@app.after_shutdown
async def disconnect(runner: Runner) -> None:
    connections.clear()
    print(f"{runner.role}: stopped")


@app.task()
async def query() -> list[str]:
    return list(connections)


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())
    handle = await query.delay()
    print("connections:", await handle.result(timeout=5))
    worker.request_shutdown()
    await run


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

One hook for every process¶

Lifecycle hooks run in every asynk process, Worker and Scheduler alike, and receive the Runner they run in. runner.role is "worker" or "scheduler", so a hook can branch where the processes differ. Annotate the parameter Runner: a hook annotated Worker is rejected by mypy, since it would not accept a scheduler.

Startup and shutdown order¶

1. The store starts.
2. The before_startup hooks run.
3. The consumer starts (worker).
4. The after_startup hooks run.
5. The process serves.
6. The before_shutdown hooks run.
7. The worker drains and stops its consumer.
8. The after_shutdown hooks run.
9. The store stops.

Every hook can use the store.

Registration¶

Each decorator appends to an ordered list, and hooks run in registration order. Hooks may be def or async def; sync hooks run on the event loop, so keep them cheap. Registering the same hook object twice is a no-op; hooks that merely share a name, such as lambdas or functions made by a factory, are all registered. A module that is imported again (for example reloaded) registers its hooks again, since its functions are new objects.

Errors¶

Hooks fail loudly. A lifecycle hook that raises makes Worker.run() or Scheduler.run() raise, and the remaining hooks of that event do not run, with one exception: every after_shutdown hook runs even when one raises. Each failure is logged. One failure is re-raised as is; several are raised together as an ExceptionGroup, which you catch with except*. Once started, the process still cleans up first, and a worker still drains its running tasks.

If a before_startup hook raises, the consumer fails to start, or run() is cancelled before startup completes, no further hook runs: not after_startup, before_shutdown or after_shutdown. Only the store is stopped, and the error propagates. If the store's stop() raises while this unwinds, that error is logged and never hides the original one.

So a resource opened by an earlier before_startup hook is not closed by your after_shutdown hook when a later step fails. Nothing closes it for you. If you call run() yourself, wrap it in try/finally. Under the asynk CLI the process exits on the error. A failure inside your own hook is yours to clean up before you re-raise.

A failing after_startup hook still runs the shutdown path, including after_shutdown (but not before_shutdown).

In a scheduler, keep hooks quick: the liveness probe is not refreshed while a hook runs, and a slow shutdown hook delays exit by its full duration. See Scheduled tasks.