lifespan

Description

Lifespan is a lifecycle manager for background tasks in Rapidy.

It manages tasks that should be started before or after the server starts or run continuously.

on_startup

on_startup refers to tasks that are executed in the event loop immediately after the application starts, along with the request handler.

on_startup handlers must have the following signature.

from rapidy import Rapidy

def startup() -> None:
    print('startup')

rapidy = Rapidy(on_startup=[startup])

from rapidy import Rapidy

def startup(rapidy: Rapidy) -> None:
    print(f'startup, application: {app}')

rapidy = Rapidy(on_startup=[startup])

from rapidy import Rapidy

async def async_startup() -> None:
    print('async_startup')

rapidy = Rapidy(on_startup=[async_startup])

from rapidy import Rapidy

async def async_startup(rapidy: Rapidy) -> None:
    print(f'async_startup, application: {app}')

rapidy = Rapidy(on_startup=[async_startup])

If a handler has attributes, the current instance of Rapidy will always be passed as the first argument.

Adding handlers to an already created Rapidy instance.

from rapidy import Rapidy

async def startup() -> None:
    print('startup')

rapidy = Rapidy()
rapidy.lifespan.on_startup.append(startup)

---

on_shutdown

on_shutdown refers to tasks executed after the server stops.

This mechanism can be used to properly close long-lived connections such as WebSockets or streaming data.

on_shutdown handlers must have the following signature.

from rapidy import Rapidy

def shutdown() -> None:
    print('shutdown')

rapidy = Rapidy(on_shutdown=[shutdown])

from rapidy import Rapidy

def shutdown(rapidy: Rapidy) -> None:
    print(f'shutdown, application: {rapidy}')

rapidy = Rapidy(on_shutdown=[shutdown])

from rapidy import Rapidy

async def async_shutdown() -> None:
    print('async_shutdown')

rapidy = Rapidy(on_shutdown=[async_shutdown])

from rapidy import Rapidy

async def async_shutdown(rapidy: Rapidy) -> None:
    print(f'async_shutdown, application: {rapidy}')

rapidy = Rapidy(on_shutdown=[async_shutdown])

If a handler has attributes, the current instance of Application will always be passed as the first argument.

Adding handlers to an already created Application instance.

from rapidy import Rapidy

async def shutdown() -> None:
    print('shutdown')

rapidy = Rapidy()
rapidy.lifespan.on_shutdown.append(shutdown)

---

on_cleanup

on_cleanup refers to tasks executed after the server stops and all on_shutdown handlers have been completed.

This signal can be used, for example, to properly close database connections.

on_cleanup handlers must have the following signature.

from rapidy import Rapidy

def cleanup() -> None:
    print('cleanup')

rapidy = Rapidy(on_cleanup=[cleanup])

from rapidy import Rapidy

def cleanup(rapidy: Rapidy) -> None:
    print(f'cleanup, application: {rapidy}')

rapidy = Rapidy(on_cleanup=[cleanup])

from rapidy import Rapidy

async def async_cleanup() -> None:
    print('async_cleanup')

rapidy = Rapidy(on_cleanup=[async_cleanup])

from rapidy import Rapidy

async def async_cleanup(rapidy: Rapidy) -> None:
    print(f'async_cleanup, application: {rapidy}')

rapidy = Rapidy(on_cleanup=[async_cleanup])

If a handler has attributes, the current instance of Application will always be passed as the first argument.

Adding handlers to an already created Application instance.

from rapidy import Rapidy

async def cleanup() -> None:
    print('cleanup')

rapidy = Rapidy()
rapidy.lifespan.on_cleanup.append(cleanup)

on_cleanup signals are executed last.

---

lifespan

lifespan is responsible for managing background tasks.

This mechanism is useful when working with long-running tasks or when it is necessary to maintain a specific context object, such as connections.

lifespan handlers must have the following signature.

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from rapidy import Rapidy

@asynccontextmanager
async def bg_task() -> AsyncGenerator[None, None]:
    try:
        print('starting background task')
        yield
    finally:
        print('finishing background task')

rapidy = Rapidy(
    lifespan=[bg_task()],
)

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from rapidy import Rapidy

@asynccontextmanager
async def bg_task_with_app(rapidy: Rapidy) -> AsyncGenerator[None, None]:
    try:
        print('starting background task')
        yield
    finally:
        print('finishing background task')

rapidy = Rapidy(
    lifespan=[bg_task_with_app],
)

Adding handlers to an already created Application instance.

from contextlib import asynccontextmanager
from typing import AsyncGenerator

from rapidy import Rapidy

@asynccontextmanager
async def bg_task() -> AsyncGenerator[None, None]:
    try:
        print('starting background task')
        yield
    finally:
        print('finishing background task')

rapidy = Rapidy()
rapidy.lifespan.append(bg_task())

lifespan tasks complete before on_cleanup handlers are executed.