INNER CODE UNIT · Python

_stop_task_queue_observability_poller

skygazer42/MimirQ · app/main.py:398

async def _stop_task_queue_observability_poller() -> None:
    if not bool(getattr(settings, "PROMETHEUS_ENABLED", False)):
        return
    try:
        from app.services.task_queue_observability_service import stop_task_queue_observability_poller

        await stop_task_queue_observability_poller()
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to stop task queue observability poller: %s", str(exc)[:200])


def _dispose_database_engine() -> None:
    try:
        engine.dispose()
        logger.info("Database engine disposed")
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to dispose database engine: %s", str(exc)[:200])

View source record →