app/utils.py

"""app/utils.py — Shared async helpers."""
from __future__ import annotations

import asyncio
import logging
from typing import Coroutine, Any

logger = logging.getLogger(__name__)


def fire_and_forget(coro: Coroutine[Any, Any, Any]) -> None:
    """Schedule a coroutine as a non-blocking background task.

    Exceptions inside the coroutine are logged but never propagate —
    a storage failure must never break a security response.
    """
    async def _guarded() -> None:
        try:
            await coro
        except Exception as exc:
            logger.warning("Background DB write failed (non-critical): %s", exc)

    try:
        asyncio.get_running_loop().create_task(_guarded())
    except RuntimeError:
        # No running event loop (e.g., pure-sync unit tests) — skip silently.
        pass