refactor: extract shared DB helpers to tasks/db.py
Extract identical get_async_session() and cleanup_engine() functions from downloads.py and maintenance.py into a shared db.py module. This eliminates code duplication and improves maintainability. db.py imports only from app.config and SQLAlchemy to avoid circular dependencies with the task modules that import from it. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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()
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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']:
|
||||
|
||||
Reference in New Issue
Block a user