"""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