run_with_asyncio

safir.asyncio.run_with_asyncio(f)

Run the decorated function with asyncio.run.

Intended to be used as a decorator around an async function that needs to be run in a sync context. The decorated function will be run with asyncio.run when invoked. The caller must not already be inside an asyncio task.

Parameters:

f (Callable[[ParamSpec(P, bound= None)], Coroutine[None, None, TypeVar(F)]]) – The function to wrap.

Return type:

Callable[[ParamSpec(P, bound= None)], TypeVar(F)]

Examples

An application that uses Safir and Click may use the following Click command function to initialize a database.

import structlog
from safir.asyncio import run_with_asyncio
from safir.database import initialize_database

from .config import config
from .schema import Base

@main.command()
@run_with_asyncio
async def init() -> None:
    logger = structlog.get_logger(config.safir.logger_name)
    engine = await initialize_database(
        config.database_url,
        config.database_password,
        logger,
        schema=Base.metadata,
    )
    await engine.dispose()