feat(fc3c): download_source Celery task + register in celery_app
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -29,6 +29,7 @@ def make_celery() -> Celery:
|
|||||||
"backend.app.tasks.thumbnail",
|
"backend.app.tasks.thumbnail",
|
||||||
"backend.app.tasks.maintenance",
|
"backend.app.tasks.maintenance",
|
||||||
"backend.app.tasks.ml",
|
"backend.app.tasks.ml",
|
||||||
|
"backend.app.tasks.download",
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
app.conf.update(
|
app.conf.update(
|
||||||
|
|||||||
@@ -0,0 +1,76 @@
|
|||||||
|
"""download_source Celery task — runs DownloadService for one source."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
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 ImportSettings
|
||||||
|
from ..services.credential_crypto import CredentialCrypto
|
||||||
|
from ..services.credential_service import CredentialService
|
||||||
|
from ..services.download_service import DownloadService
|
||||||
|
from ..services.gallery_dl import GalleryDLService
|
||||||
|
from ..services.importer import Importer
|
||||||
|
from ..services.thumbnailer import Thumbnailer
|
||||||
|
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, acks_late=True)
|
||||||
|
def download_source(self, source_id: int) -> int:
|
||||||
|
"""Returns the DownloadEvent.id."""
|
||||||
|
|
||||||
|
async def _run():
|
||||||
|
async_factory, async_engine = _async_session_factory()
|
||||||
|
SyncFactory = _sync_session_factory()
|
||||||
|
try:
|
||||||
|
with SyncFactory() as sync_session:
|
||||||
|
settings = sync_session.execute(
|
||||||
|
select(ImportSettings).where(ImportSettings.id == 1)
|
||||||
|
).scalar_one()
|
||||||
|
rate_limit = settings.download_rate_limit_seconds
|
||||||
|
validate_files = settings.download_validate_files
|
||||||
|
|
||||||
|
gdl = GalleryDLService(
|
||||||
|
images_root=IMAGES_ROOT,
|
||||||
|
rate_limit=rate_limit,
|
||||||
|
validate_files=validate_files,
|
||||||
|
)
|
||||||
|
crypto = CredentialCrypto(_KEY_PATH)
|
||||||
|
|
||||||
|
async with async_factory() as async_session:
|
||||||
|
cred_service = CredentialService(async_session, crypto)
|
||||||
|
with SyncFactory() as sync_session:
|
||||||
|
sync_settings = sync_session.execute(
|
||||||
|
select(ImportSettings).where(ImportSettings.id == 1)
|
||||||
|
).scalar_one()
|
||||||
|
importer = Importer(
|
||||||
|
session=sync_session,
|
||||||
|
images_root=IMAGES_ROOT,
|
||||||
|
import_root=IMAGES_ROOT,
|
||||||
|
thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
|
||||||
|
settings=sync_settings,
|
||||||
|
)
|
||||||
|
svc = DownloadService(
|
||||||
|
async_session=async_session,
|
||||||
|
sync_session=sync_session,
|
||||||
|
gdl=gdl,
|
||||||
|
importer=importer,
|
||||||
|
cred_service=cred_service,
|
||||||
|
)
|
||||||
|
return await svc.download_source(source_id)
|
||||||
|
finally:
|
||||||
|
await async_engine.dispose()
|
||||||
|
|
||||||
|
return asyncio.run(_run())
|
||||||
@@ -0,0 +1,14 @@
|
|||||||
|
"""Smoke test that the Celery task is registered and routed correctly."""
|
||||||
|
|
||||||
|
from backend.app.celery_app import celery
|
||||||
|
|
||||||
|
|
||||||
|
def test_download_source_is_registered():
|
||||||
|
name = "backend.app.tasks.download.download_source"
|
||||||
|
assert name in celery.tasks
|
||||||
|
|
||||||
|
|
||||||
|
def test_download_source_routes_to_download_queue():
|
||||||
|
routes = celery.conf.task_routes
|
||||||
|
assert "backend.app.tasks.download.*" in routes
|
||||||
|
assert routes["backend.app.tasks.download.*"]["queue"] == "download"
|
||||||
Reference in New Issue
Block a user