Document task DB access and unify background task DB handling

backend/app/tasks/blocklist_import.py

@@ -17,12 +17,11 @@ from typing import TYPE_CHECKING, Any
 
 import structlog
 
-from app.db import open_db
 from app.services import ban_service, blocklist_service
+from app.tasks.db import task_db
 from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
-    import aiosqlite
     from aiohttp import ClientSession
     from fastapi import FastAPI
 
@@ -34,11 +33,6 @@ log: structlog.stdlib.BoundLogger = structlog.get_logger()
 JOB_ID: str = "blocklist_import"
 
 
-async def _get_db(settings: Settings) -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(settings.database_path)
-    return db, True
-
-
 async def _run_import_with_resources(settings: Settings, http_session: ClientSession) -> None:
     """APScheduler callback that imports all enabled blocklist sources.
 
@@ -46,17 +40,17 @@ async def _run_import_with_resources(settings: Settings, http_session: ClientSes
         settings: The resolved application settings used for database access.
         http_session: The shared aiohttp session used for blocklist downloads.
     """
-    db, close_db = await _get_db(settings)
     socket_path: str = settings.fail2ban_socket
 
     log.info("blocklist_import_starting")
     try:
-        result = await blocklist_service.import_all(
-            db,
-            http_session,
-            socket_path,
-            ban_ip=ban_service.ban_ip,
-        )
+        async with task_db(settings) as db:
+            result = await blocklist_service.import_all(
+                db,
+                http_session,
+                socket_path,
+                ban_ip=ban_service.ban_ip,
+            )
         log.info(
             "blocklist_import_finished",
             total_imported=result.total_imported,
@@ -65,9 +59,6 @@ async def _run_import_with_resources(settings: Settings, http_session: ClientSes
         )
     except Exception:
         log.exception("blocklist_import_unexpected_error")
-    finally:
-        if close_db:
-            await db.close()
 
 
 run_import_with_resources = _run_import_with_resources
@@ -91,12 +82,8 @@ async def register(app: FastAPI) -> None:
             ``app.state.scheduler`` will receive the job.
     """
     settings = get_effective_settings(app)
-    db, close_db = await _get_db(settings)
-    try:
+    async with task_db(settings) as db:
         config = await blocklist_service.get_schedule(db)
-    finally:
-        if close_db:
-            await db.close()
 
     _apply_schedule(app, config)
 
@@ -114,12 +101,8 @@ def reschedule(app: FastAPI) -> None:
 
     async def _do_reschedule() -> None:
         settings = get_effective_settings(app)
-        db, close_db = await _get_db(settings)
-        try:
+        async with task_db(settings) as db:
             config = await blocklist_service.get_schedule(db)
-        finally:
-            if close_db:
-                await db.close()
         _apply_schedule(app, config)
 
     asyncio.ensure_future(_do_reschedule())

backend/app/tasks/db.py

@@ -0,0 +1,29 @@
+"""Shared database helpers for APScheduler background tasks."""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
+
+from app.db import open_db
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    import aiosqlite
+
+    from app.config import Settings
+
+
+@asynccontextmanager
+async def task_db(settings: Settings) -> AsyncIterator[aiosqlite.Connection]:
+    """Open a dedicated application database connection for a background task.
+
+    Background tasks run outside FastAPI request scope and therefore must
+    manage their own SQLite connection instead of using FastAPI dependencies.
+    """
+    db = await open_db(settings.database_path)
+    try:
+        yield db
+    finally:
+        await db.close()

backend/app/tasks/geo_cache_flush.py

@@ -11,21 +11,19 @@ at risk on an unexpected process restart.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import structlog
 
-from app.db import open_db
+from app.services import geo_service
+from app.tasks.db import task_db
 from app.utils.runtime_state import get_effective_settings
 
-if TYPE_CHECKING:
-    import aiosqlite
-    from app.config import Settings
-from app.services import geo_service
-
 if TYPE_CHECKING:
     from fastapi import FastAPI
 
+    from app.config import Settings
+
 log: structlog.stdlib.BoundLogger = structlog.get_logger()
 
 #: How often the flush job fires (seconds).  Configurable tuning constant.
@@ -35,23 +33,14 @@ GEO_FLUSH_INTERVAL: int = 60
 JOB_ID: str = "geo_cache_flush"
 
 
-async def _get_db(settings: "Settings") -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(settings.database_path)
-    return db, True
-
-
-async def _run_flush_with_settings(settings: "Settings") -> None:
+async def _run_flush_with_settings(settings: Settings) -> None:
     """Flush the geo service dirty set to the application database.
 
     Args:
         settings: The resolved application settings used for database access.
     """
-    db, close_db = await _get_db(settings)
-    try:
+    async with task_db(settings) as db:
         count = await geo_service.flush_dirty(db)
-    finally:
-        if close_db:
-            await db.close()
 
     if count > 0:
         log.debug("geo_cache_flush_ran", flushed=count)

backend/app/tasks/geo_re_resolve.py

@@ -21,18 +21,16 @@ from typing import TYPE_CHECKING
 
 import structlog
 
-from app.db import open_db
+from app.services import geo_service
+from app.tasks.db import task_db
 from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
-    import aiosqlite
     from aiohttp import ClientSession
-    from app.config import Settings
-from app.services import geo_service
-
-if TYPE_CHECKING:
     from fastapi import FastAPI
 
+    from app.config import Settings
+
 log: structlog.stdlib.BoundLogger = structlog.get_logger()
 
 #: How often the re-resolve job fires (seconds).  10 minutes.
@@ -42,21 +40,14 @@ GEO_RE_RESOLVE_INTERVAL: int = 600
 JOB_ID: str = "geo_re_resolve"
 
 
-async def _get_db(settings: "Settings") -> tuple[aiosqlite.Connection, bool]:
-    db = await open_db(settings.database_path)
-    return db, True
-
-
-async def _run_re_resolve_with_resources(settings: "Settings", http_session: "ClientSession") -> None:
+async def _run_re_resolve_with_resources(settings: Settings, http_session: ClientSession) -> None:
     """Query NULL-country IPs from the database and re-resolve them.
 
     Args:
         settings: The resolved application settings used for database access.
         http_session: The shared aiohttp session used for external lookups.
     """
-    db, close_db = await _get_db(settings)
-
-    try:
+    async with task_db(settings) as db:
         # Fetch all IPs with NULL country_code from the persistent cache.
         unresolved_ips = await geo_service.get_unresolved_ips(db)
 
@@ -73,17 +64,14 @@ async def _run_re_resolve_with_resources(settings: "Settings", http_session: "Cl
         # passed.  This is a background task so DB writes are allowed.
         results = await geo_service.lookup_batch(unresolved_ips, http_session, db=db)
 
-        resolved_count: int = sum(
-            1 for info in results.values() if info.country_code is not None
-        )
-        log.info(
-            "geo_re_resolve_complete",
-            retried=len(unresolved_ips),
-            resolved=resolved_count,
-        )
-    finally:
-        if close_db:
-            await db.close()
+    resolved_count: int = sum(
+        1 for info in results.values() if info.country_code is not None
+    )
+    log.info(
+        "geo_re_resolve_complete",
+        retried=len(unresolved_ips),
+        resolved=resolved_count,
+    )
 
 
 async def _run_re_resolve(app: FastAPI) -> None:

backend/app/tasks/history_sync.py

@@ -11,8 +11,8 @@ from typing import TYPE_CHECKING
 
 import structlog
 
-from app.db import open_db
 from app.services import history_service
+from app.tasks.db import task_db
 from app.utils.runtime_state import get_effective_settings
 
 if TYPE_CHECKING:
@@ -34,15 +34,13 @@ BACKFILL_WINDOW: int = 648000
 
 async def _run_sync_with_settings(settings: Settings) -> None:
     socket_path: str = settings.fail2ban_socket
-    db = await open_db(settings.database_path)
 
     try:
-        synced = await history_service.sync_from_fail2ban_db(db, socket_path)
+        async with task_db(settings) as db:
+            synced = await history_service.sync_from_fail2ban_db(db, socket_path)
         log.info("history_sync_complete", synced=synced)
     except Exception:
         log.exception("history_sync_failed")
-    finally:
-        await db.close()
 
 
 async def _run_sync(app: FastAPI) -> None: