● Zion Boggan
repos/Sightline/backend/app/workers/tasks.py
zionboggan.com ↗
52 lines · python
History for this file →
1
from __future__ import annotations
2
 
3
from app.db.session import SessionLocal
4
from app.workers.celery_app import celery_app
5
 
6
 
7
@celery_app.task(name="sightline.process_import")
8
def process_import_task(import_id: str) -> dict:
9
    from app.services.ingest import process_import
10
    db = SessionLocal()
11
    try:
12
        return process_import(db, import_id)
13
    finally:
14
        db.close()
15
 
16
 
17
@celery_app.task(name="sightline.run_enrichment")
18
def run_enrichment_task(job_id: str) -> dict:
19
    """Background enrichment for slow/network-bound enrichers."""
20
    from datetime import datetime, timezone
21
    from app.models import EnrichmentJob
22
    from app.services.enrichment import run_enricher
23
    db = SessionLocal()
24
    try:
25
        job = db.get(EnrichmentJob, job_id)
26
        if not job:
27
            return {"error": "job not found"}
28
        result = run_enricher(job.enricher, job.target)
29
        job.result = result.to_dict()
30
        job.confidence = result.confidence
31
        job.status = "done" if not result.error else "error"
32
        job.finished_at = datetime.now(timezone.utc)
33
        db.commit()
34
        return result.to_dict()
35
    finally:
36
        db.close()
37
 
38
 
39
@celery_app.task(name="sightline.apply_retention")
40
def apply_retention_task() -> dict:
41
    """Scheduled retention enforcement - delete reads older than the configured window."""
42
    from datetime import datetime, timedelta, timezone
43
    from app.core.config import settings
44
    from app.services.retention import purge_expired
45
    db = SessionLocal()
46
    try:
47
        cutoff = datetime.now(timezone.utc) - timedelta(days=settings.default_retention_days)
48
        deleted = purge_expired(db, cutoff)
49
        db.commit()
50
        return {"deleted": deleted, "cutoff": cutoff.isoformat()}
51
    finally:
52
        db.close()