lifespan

Описание

Lifespan — это менеджер жизненного цикла фоновых задач в Rapidy.

Он управляет задачами, которые должны быть запущены до или после старта сервера, либо выполняться постоянно.

on_startup

on_startup — это задачи, которые запускаются в цикле событий сразу после старта приложения вместе с обработчиком запросов.

Обработчики on_startup должны иметь следующую сигнатуру.

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])

Если у обработчика есть атрибуты, первым всегда будет передаваться текущий экземпляр Rapidy.

Добавление обработчиков в уже созданный объект Rapidy.

from rapidy import Rapidy

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

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

---

on_shutdown

on_shutdown — это задачи, выполняемые после остановки сервера.

Этот механизм можно использовать для корректного завершения долгоживущих соединений, таких как веб-сокеты или потоковая передача данных.

Обработчики on_shutdown должны иметь следующую сигнатуру.

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])

Если у обработчика есть атрибуты, первым всегда будет передаваться текущий экземпляр Application.

Добавление обработчиков в уже созданный объект Application.

from rapidy import Rapidy

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

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

---

on_cleanup

on_cleanup — это задачи, запускаемые после остановки сервера и выполнения всех on_shutdown обработчиков.

Этот сигнал можно использовать, например, для корректного закрытия соединений с базой данных.

Обработчики on_cleanup должны иметь следующую сигнатуру.

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])

Если у обработчика есть атрибуты, первым всегда будет передаваться текущий экземпляр Application.

Добавление обработчиков в уже созданный объект Application.

from rapidy import Rapidy

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

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

Сигналы on_cleanup выполняются последними.

---

lifespan

lifespan отвечает за управление фоновыми задачами.

Этот механизм полезен при работе с долгоживущими задачами или в случаях, когда необходимо поддерживать определённый объект контекста, например соединения.

Обработчики lifespan должны иметь следующую сигнатуру.

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],
)

Добавление обработчиков в уже созданный объект Application.

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 завершаются перед выполнением обработчиков on_cleanup.