diff --git a/backend/app/tasks/db.py b/backend/app/tasks/db.py
new file mode 100644
index 0000000..50cb612
--- /dev/null
+++ b/backend/app/tasks/db.py
@@ -0,0 +1,27 @@
+"""Shared async database helpers for Celery tasks.
+
+Provides get_async_session() and cleanup_engine() used by both
+downloads.py and maintenance.py. This module must not import from
+either of those files to avoid circular imports.
+"""
+
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+
+from app.config import get_settings
+
+
+def get_async_session():
+    """Get async session factory for Celery tasks.
+
+    Creates a fresh engine and session factory each time to avoid
+    event loop issues when asyncio.run() creates new loops.
+    """
+    settings = get_settings()
+    engine = create_async_engine(settings.async_database_url, echo=False)
+    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    return session_factory, engine
+
+
+async def cleanup_engine(engine):
+    """Properly dispose of the async engine."""
+    await engine.dispose()
diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py
index 235963b..08d4957 100644
--- a/backend/app/tasks/downloads.py
+++ b/backend/app/tasks/downloads.py
@@ -6,10 +6,9 @@ from datetime import datetime, timedelta
 from typing import Optional
 
 from sqlalchemy import select, and_, or_, update
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import selectinload
 
-from app.config import get_settings
+from app.tasks.db import get_async_session, cleanup_engine as _cleanup_engine
 from app.tasks.celery_app import celery_app
 from app.models.base import utcnow
 from app.models.source import Source
@@ -23,25 +22,10 @@ from app.events import publish_event
 logger = logging.getLogger(__name__)
 
 # Settings for database connection
+from app.config import get_settings
 settings = get_settings()
 
 
-def get_async_session():
-    """Get async session factory for Celery tasks.
-
-    Creates a fresh engine and session factory each time to avoid
-    event loop issues when asyncio.run() creates new loops.
-    """
-    engine = create_async_engine(settings.async_database_url, echo=False)
-    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-    return session_factory, engine
-
-
-async def _cleanup_engine(engine):
-    """Properly dispose of the async engine."""
-    await engine.dispose()
-
-
 async def _get_db_setting(session: AsyncSession, key: str, default=None):
     """Get a setting value from the database, falling back to defaults.
 
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index 2faaa63..ca50073 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -7,9 +7,8 @@ from datetime import datetime, timedelta
 from pathlib import Path
 
 from sqlalchemy import select, update
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 
-from app.config import get_settings
+from app.tasks.db import get_async_session, cleanup_engine as _cleanup_engine
 from app.tasks.celery_app import celery_app
 from app.models.setting import Setting, STORAGE_STATS_SETTING
 from app.models.download import Download, DownloadStatus
@@ -17,6 +16,7 @@ from app.models.base import utcnow
 
 logger = logging.getLogger(__name__)
 
+from app.config import get_settings
 settings = get_settings()
 
 # Maximum time a job can be in "running" state before considered orphaned
@@ -29,18 +29,6 @@ STALE_RUNNING_THRESHOLD_MINUTES = 90
 STALE_QUEUED_THRESHOLD_MINUTES = 30
 
 
-def get_async_session():
-    """Get async session factory for Celery tasks."""
-    engine = create_async_engine(settings.async_database_url, echo=False)
-    session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-    return session_factory, engine
-
-
-async def _cleanup_engine(engine):
-    """Properly dispose of the async engine."""
-    await engine.dispose()
-
-
 def format_size(size_bytes: int) -> str:
     """Format bytes as human-readable size."""
     for unit in ['B', 'KB', 'MB', 'GB', 'TB']: