diff --git a/backend/app/tasks/_async_session.py b/backend/app/tasks/_async_session.py new file mode 100644 index 0000000..ea7a796 --- /dev/null +++ b/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 diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index 8801a56..fa82212 100644 --- a/backend/app/tasks/download.py +++ b/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: diff --git a/backend/app/tasks/migration.py b/backend/app/tasks/migration.py index a59c8b2..aec5102 100644 --- a/backend/app/tasks/migration.py +++ b/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") diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 60cbfe0..c5f3dff 100644 --- a/backend/app/tasks/scan.py +++ b/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).