refactor(dry-B4): extract shared async_session_factory for Celery tasks

download/migration/scan each defined an identical _async_session_factory()
(fresh per-invocation async engine — async connections are event-loop-bound
so each asyncio.run() task needs its own engine, unlike the process-wide
_sync_engine). Moved it to tasks/_async_session.py; the 3 files import it
and drop their now-orphaned sqlalchemy.ext.asyncio / get_config imports
(migration keeps AsyncSession for a type hint). Call-site try/finally
dispose left as-is to avoid re-indenting the critical task bodies.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/tasks/_async_session.py

@@ -0,0 +1,21 @@
+"""Per-invocation async session factory for Celery task modules.
+
+Async engine connections are bound to the event loop. Each Celery task
+runs its async body under a fresh ``asyncio.run()`` loop, so it needs its
+own engine created (and disposed) within that loop — a process-wide async
+engine would reuse loop-bound connections across tasks and raise "attached
+to a different loop". So unlike the process-wide sync engine in
+``_sync_engine.py``, this returns a fresh engine per call; the caller
+disposes it (``await engine.dispose()``) when its loop ends.
+"""
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from ..config import get_config
+
+
+def async_session_factory():
+    """Return ``(sessionmaker, engine)`` bound to a fresh async engine."""
+    cfg = get_config()
+    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
+    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine

backend/app/tasks/download.py

@@ -5,10 +5,8 @@ from pathlib import Path
 
 from sqlalchemy import select
 from sqlalchemy.exc import DBAPIError, OperationalError
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from ..celery_app import celery
-from ..config import get_config
 from ..models import ImportSettings
 from ..services.credential_crypto import CredentialCrypto
 from ..services.credential_service import CredentialService
@@ -16,18 +14,13 @@ from ..services.download_service import DownloadService
 from ..services.gallery_dl import GalleryDLService
 from ..services.importer import Importer
 from ..services.thumbnailer import Thumbnailer
+from ._async_session import async_session_factory
 from .import_file import _sync_session_factory
 
 IMAGES_ROOT = Path("/images")
 _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
 
 
-def _async_session_factory():
-    cfg = get_config()
-    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
-    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine
-
-
 @celery.task(
     name="backend.app.tasks.download.download_source",
     bind=True,
@@ -44,7 +37,7 @@ def download_source(self, source_id: int) -> int:
     """Returns the DownloadEvent.id."""
 
     async def _run():
-        async_factory, async_engine = _async_session_factory()
+        async_factory, async_engine = async_session_factory()
         SyncFactory = _sync_session_factory()
         try:
             with SyncFactory() as sync_session:

backend/app/tasks/migration.py

@@ -15,14 +15,14 @@ from datetime import UTC, datetime
 from pathlib import Path
 
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..celery_app import celery
-from ..config import get_config
 from ..models import MigrationRun
 from ..services.credential_crypto import CredentialCrypto
 from ..services.migrators import cleanup as cleanup_mod
 from ..services.migrators import gs_ingest, ir_ingest, ml_queue, tag_apply, verify
+from ._async_session import async_session_factory
 
 log = logging.getLogger(__name__)
 
@@ -30,12 +30,6 @@ IMAGES_ROOT = Path("/images")
 _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
 
 
-def _async_session_factory():
-    cfg = get_config()
-    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
-    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine
-
-
 async def _update_run(
     db: AsyncSession, run_id: int, *,
     status: str | None = None, counts: dict | None = None,
@@ -59,7 +53,7 @@ async def _update_run(
 
 
 async def _run_async(run_id: int, kind: str, params: dict) -> dict:
-    factory, engine = _async_session_factory()
+    factory, engine = async_session_factory()
     try:
         async with factory() as db:
             await _update_run(db, run_id, status="running")

backend/app/tasks/scan.py

@@ -12,13 +12,12 @@ from datetime import UTC, datetime
 from pathlib import Path
 
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from ..celery_app import celery
-from ..config import get_config
 from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
 from ..services.archive_extractor import is_archive
 from ..services.scheduler_service import record_tick, select_due_sources
+from ._async_session import async_session_factory
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
 
@@ -141,14 +140,8 @@ def scan_directory(self, triggered_by: str = "manual",
 # --- FC-3d: periodic source-check tick ------------------------------------
 
 
-def _async_session_factory():
-    cfg = get_config()
-    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
-    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine
-
-
 async def _tick_due_sources_async() -> dict:
-    factory, engine = _async_session_factory()
+    factory, engine = async_session_factory()
     try:
         async with factory() as session:
             # Prove the scheduler is alive even on empty ticks (UI reads this).