bvandeusen
b4e0d680f1
Ruff lint surfaced 23 violations across three rules; all addressed: UP017 (Use datetime.UTC alias): Replaced 13 sites of datetime.now(timezone.utc) with datetime.now(UTC), also adjusted from-imports accordingly. UTC is a Python 3.11+ alias for timezone.utc that ruff's pyupgrade rules prefer. UP042 (StrEnum): Replaced `class TagKind(str, Enum)` and `class SkipReason(str, Enum)` with `class Foo(StrEnum)`. StrEnum was added in Python 3.11 stdlib and is the modern idiom. Behavior is equivalent for our usage (the .value attribute, str(member) semantics). I001 (Import sorting): Added `known-first-party = ["backend"]` to ruff.toml's [lint.isort] so ruff groups `backend.*` imports correctly. Without it, ruff treated them as third-party and demanded a different grouping. The existing import order is stdlib → third-party → first-party → local relative, which ruff now accepts. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
"""Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy import create_engine, delete, select, update
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from ..celery_app import celery
|
|
from ..config import get_config
|
|
from ..models import ImportTask
|
|
|
|
|
|
STUCK_THRESHOLD_MINUTES = 30
|
|
OLD_TASK_DAYS = 7
|
|
|
|
|
|
def _sync_session_factory():
|
|
cfg = get_config()
|
|
engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True)
|
|
return sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
|
|
def recover_interrupted_tasks() -> int:
|
|
"""Find ImportTask rows stuck in 'processing' for >30 min and re-queue them.
|
|
|
|
Why 30 min: large videos can legitimately take many minutes to import;
|
|
30 is a safe gate that catches actual crashes (which leave the row stuck
|
|
forever) without resetting slow-but-still-running jobs.
|
|
"""
|
|
SessionLocal = _sync_session_factory()
|
|
cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
|
|
with SessionLocal() as session:
|
|
stuck_ids = session.execute(
|
|
select(ImportTask.id)
|
|
.where(ImportTask.status == "processing")
|
|
.where(ImportTask.started_at < cutoff)
|
|
).scalars().all()
|
|
|
|
if not stuck_ids:
|
|
return 0
|
|
|
|
session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.id.in_(stuck_ids))
|
|
.values(status="queued", started_at=None, error="recovered from stuck state")
|
|
)
|
|
session.commit()
|
|
|
|
from .import_file import import_media_file
|
|
for tid in stuck_ids:
|
|
import_media_file.delay(tid)
|
|
|
|
return len(stuck_ids)
|
|
|
|
|
|
@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
|
|
def cleanup_old_tasks() -> int:
|
|
"""Delete completed/skipped/failed ImportTask rows older than 7 days.
|
|
|
|
Why 7 days: long enough to debug an issue an operator only notices days
|
|
later; short enough that the task table stays a useful operational view
|
|
rather than an archive. Matches IR's default.
|
|
"""
|
|
SessionLocal = _sync_session_factory()
|
|
cutoff = datetime.now(UTC) - timedelta(days=OLD_TASK_DAYS)
|
|
with SessionLocal() as session:
|
|
result = session.execute(
|
|
delete(ImportTask)
|
|
.where(ImportTask.status.in_(["complete", "skipped", "failed"]))
|
|
.where(ImportTask.finished_at < cutoff)
|
|
)
|
|
session.commit()
|
|
return result.rowcount or 0
|