CultivaHub API — Patrones FastAPI Producción

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.routers import auth, agents, credits

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()  # cleanup pool

def create_app() -> FastAPI:
    app = FastAPI(
        title="CultivaHub API",
        version="1.0.0",
        lifespan=lifespan,
    )
    app.add_middleware(CORSMiddleware,
        allow_origins=settings.allowed_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    app.include_router(auth.router,
        prefix="/auth", tags=["auth"])
    app.include_router(agents.router,
        prefix="/agents", tags=["agents"])
    app.include_router(credits.router,
        prefix="/credits", tags=["credits"])
    return app

from datetime import datetime
from pydantic import BaseModel, Field, model_validator
from enum import StrEnum

class AgentType(StrEnum):
    CHAT     = "chat"
    SEARCH   = "search"
    PIPELINE = "pipeline"

class AgentCreate(BaseModel):
    name: str = Field(min_length=3, max_length=80)
    agent_type: AgentType
    llm_model: str = "claude-3-5-sonnet-20241022"
    config: dict = Field(default_factory=dict)
    credits_per_run: int = Field(ge=1, le=1000)

class AgentUpdate(BaseModel):
    name: str | None = None
    llm_model: str | None = None
    config: dict | None = None
    is_active: bool | None = None

class AgentResponse(BaseModel):
    id: int
    name: str
    agent_type: AgentType
    llm_model: str
    is_active: bool
    credits_per_run: int
    created_at: datetime
    model_config = {"from_attributes": True}

class AgentListResponse(BaseModel):
    total: int; items: list[AgentResponse]

from typing import Annotated, AsyncGenerator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import JWTError, jwt

oauth2 = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback(); raise

async def get_current_agency(
    token: Annotated[str, Depends(oauth2)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> Agency:
    exc = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token,
            settings.secret_key,
            algorithms=[settings.algorithm])
        agency_id = int(payload["sub"])
    except (JWTError, TypeError, ValueError):
        raise exc
    agency = await db.get(Agency, agency_id)
    if not agency: raise exc
    return agency

# Type aliases → routers más limpios
DbDep = Annotated[AsyncSession, Depends(get_db)]
AgencyDep = Annotated[Agency, Depends(get_current_agency)]

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.agent import Agent
from app.schemas.agent import AgentCreate, AgentUpdate

class AgentService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def create(self, agency_id: int,
                      payload: AgentCreate) -> Agent:
        agent = Agent(**payload.model_dump(),
                      agency_id=agency_id)
        self.db.add(agent)
        await self.db.commit()
        await self.db.refresh(agent)
        return agent

    async def list(self, agency_id: int,
                    skip: int = 0,
                    limit: int = 20) -> tuple:
        total = (await self.db.execute(
            select(func.count(Agent.id))
            .where(Agent.agency_id == agency_id)
        )).scalar_one()
        items = (await self.db.execute(
            select(Agent)
            .where(Agent.agency_id == agency_id)
            .order_by(Agent.id)   # orden determinista
            .offset(skip).limit(limit)
        )).scalars().all()
        return list(items), total

    async def update(self, agent_id: int,
                      agency_id: int,
                      payload: AgentUpdate) -> Agent | None:
        agent = await self.db.get(Agent, agent_id)
        if not agent or agent.agency_id != agency_id:
            return None
        for f, v in payload.model_dump(
                exclude_unset=True).items():
            setattr(agent, f, v)
        await self.db.commit()
        await self.db.refresh(agent)
        return agent