bruno
eda4faeb78
- process_image_pipeline: wrapped in try/finally to guarantee processing_status is always updated, even on unexpected crashes/timeouts - init_db: recovery query that marks images stuck in 'processing' for >10min as ERROR on startup - Clean rewrite of pipeline.py for consistent indentation
115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
"""
|
|
Configuration SQLAlchemy — session async
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlalchemy.orm import DeclarativeBase
|
|
from sqlalchemy import select, text
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
engine = create_async_engine(
|
|
settings.DATABASE_URL,
|
|
echo=settings.DEBUG,
|
|
future=True,
|
|
)
|
|
|
|
AsyncSessionLocal = async_sessionmaker(
|
|
bind=engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
autoflush=False,
|
|
autocommit=False,
|
|
)
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
async def get_db() -> AsyncSession:
|
|
"""Dependency FastAPI — injecte une session DB dans chaque requête."""
|
|
async with AsyncSessionLocal() as session:
|
|
try:
|
|
yield session
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback()
|
|
raise
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def init_db():
|
|
"""Crée toutes les tables, applique les migrations, et initialise un client par défaut si nécessaire."""
|
|
import secrets
|
|
from app.dependencies.auth import hash_api_key
|
|
from app.models.client import APIClient, ClientPlan
|
|
from app.models.image import Image, ProcessingStatus
|
|
|
|
async with engine.begin() as conn:
|
|
from app.models import image # noqa: F401
|
|
from app.models import client # noqa: F401
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Migrations ponctuelles — ajout de colonnes manquantes (PostgreSQL seulement)
|
|
try:
|
|
await conn.execute(text(
|
|
"ALTER TABLE api_clients ADD COLUMN IF NOT EXISTS ai_provider VARCHAR(64)"
|
|
))
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
await conn.execute(text(
|
|
"ALTER TABLE api_clients ADD COLUMN IF NOT EXISTS ai_model VARCHAR(256)"
|
|
))
|
|
except Exception:
|
|
pass
|
|
|
|
# Vérifier s'il y a déjà des clients + recovery des images bloquées
|
|
async with AsyncSessionLocal() as session:
|
|
# Recovery : marquer les images bloquées en "processing" depuis > 10 min comme ERROR
|
|
cutoff = datetime.now(timezone.utc) - timedelta(minutes=10)
|
|
stuck_query = select(Image).where(
|
|
Image.processing_status == ProcessingStatus.PROCESSING,
|
|
Image.processing_started_at < cutoff,
|
|
)
|
|
stuck_result = await session.execute(stuck_query)
|
|
stuck_images = stuck_result.scalars().all()
|
|
for img in stuck_images:
|
|
img.processing_status = ProcessingStatus.ERROR
|
|
img.processing_error = "[Recovery] Pipeline interrompu (timeout ou crash worker)"
|
|
img.processing_done_at = datetime.now(timezone.utc)
|
|
if stuck_images:
|
|
await session.commit()
|
|
logger.warning(
|
|
"init_db.recovery",
|
|
extra={"stuck_images_recovered": len(stuck_images)},
|
|
)
|
|
|
|
result = await session.execute(select(APIClient).limit(1))
|
|
if result.scalar_one_or_none() is None:
|
|
# Table vide -> Création du client bootstrap
|
|
raw_key = secrets.token_urlsafe(32)
|
|
key_hash = hash_api_key(raw_key)
|
|
|
|
bootstrap_client = APIClient(
|
|
name="Default Admin",
|
|
api_key_hash=key_hash,
|
|
scopes=["images:read", "images:write", "images:delete", "ai:use", "admin"],
|
|
plan=ClientPlan.PREMIUM,
|
|
)
|
|
session.add(bootstrap_client)
|
|
await session.commit()
|
|
|
|
msg = f"Bootstrap client created! ID: {bootstrap_client.id} | API_KEY: {raw_key}"
|
|
logger.info(msg, extra={
|
|
"client_id": bootstrap_client.id,
|
|
"api_key": raw_key,
|
|
"warning": "Notez cette clé ! Elle ne sera plus affichée.",
|
|
})
|