Spaces:
Build error
Build error
File size: 2,928 Bytes
e4bd765 5d9b495 e4bd765 fc1a5f0 e4bd765 fc1a5f0 e4bd765 fc1a5f0 e4bd765 5d9b495 193eb98 9097545 193eb98 9097545 5d9b495 e4bd765 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """
Exécution séquentielle du pipeline sur tous les jobs d'un corpus (Sprint 4 — Session C).
Point d'entrée : execute_corpus_job(corpus_id)
→ récupère tous les jobs PENDING du corpus
→ les exécute séquentiellement (pas de parallélisme au MVP)
→ retourne un résumé {total, done, failed}
Chaque page reçoit sa propre session pour isoler les échecs.
"""
# 1. stdlib
import asyncio
import logging
# 2. third-party
from sqlalchemy import select
# 3. local
from app.models.database import async_session_factory
from app.models.job import JobModel
logger = logging.getLogger(__name__)
async def execute_corpus_job(corpus_id: str) -> dict:
"""Lance tous les jobs PENDING du corpus séquentiellement.
Chaque job est exécuté dans sa propre session (isolement des échecs).
Un job FAILED n'interrompt pas les suivants.
Returns:
{"total": int, "done": int, "failed": int}
"""
# Collecte et verrouillage des jobs PENDING : on les passe immédiatement
# en "claimed" pour éviter qu'un second appel concurrent ne les reprenne
# (TOCTOU). Chaque job sera ensuite passé en "running" par job_runner.
async with async_session_factory() as db:
result = await db.execute(
select(JobModel).where(
JobModel.corpus_id == corpus_id,
JobModel.status == "pending",
)
)
pending_jobs = list(result.scalars().all())
job_ids: list[str] = [j.id for j in pending_jobs]
for j in pending_jobs:
j.status = "claimed"
await db.commit()
if not job_ids:
logger.info(
"Corpus run : aucun job pending",
extra={"corpus_id": corpus_id},
)
return {"total": 0, "done": 0, "failed": 0}
logger.info(
"Corpus run démarré",
extra={"corpus_id": corpus_id, "jobs": len(job_ids)},
)
# Exécution concurrente avec semaphore — chaque job gère sa propre session
from app.services.job_runner import execute_page_job
from app.config import settings
sem = asyncio.Semaphore(settings.ai_max_concurrent)
async def _run_one(jid: str) -> None:
async with sem:
await execute_page_job(jid)
await asyncio.gather(*[_run_one(jid) for jid in job_ids])
# Bilan final
async with async_session_factory() as db:
result = await db.execute(
select(JobModel).where(JobModel.id.in_(job_ids))
)
jobs = list(result.scalars().all())
done = sum(1 for j in jobs if j.status == "done")
failed = sum(1 for j in jobs if j.status == "failed")
total = len(job_ids)
logger.info(
"Corpus run terminé",
extra={
"corpus_id": corpus_id,
"total": total,
"done": done,
"failed": failed,
},
)
return {"total": total, "done": done, "failed": failed}
|