FastAPI Backend — CEI-001

Architecture Pattern

Route Layer (FastAPI)
  ↓ (validates request)
Service Layer (business logic)
  ↓ (Pydantic schema validation)
Data Layer (SQLAlchemy ORM)
  ↓
Response (Pydantic schema)

Route Patterns

Basic CRUD Route

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.schemas.evaluation import EvaluationCreate, EvaluationResponse
from app.services.evaluation_service import EvaluationService
from app.api.deps import get_db, get_current_user

router = APIRouter(prefix="/api/evaluations", tags=["evaluations"])

@router.post("", response_model=EvaluationResponse, status_code=status.HTTP_201_CREATED)
async def create_evaluation(
    eval_data: EvaluationCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
) -> EvaluationResponse:
    """
    Create new evaluation for user
    
    Example:
    ```
    POST /api/evaluations
    {
        "company_id": "123e4567-e89b-12d3-a456-426614174000",
        "project_type": "new_erp"
    }
    ```
    """
    service = EvaluationService(db)
    return await service.create(current_user.id, eval_data)

@router.get("/{eval_id}", response_model=EvaluationResponse)
async def get_evaluation(
    eval_id: str,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
) -> EvaluationResponse:
    """Get evaluation by ID"""
    service = EvaluationService(db)
    evaluation = await service.get(eval_id, current_user.id)
    if not evaluation:
        raise HTTPException(status_code=404, detail="Evaluation not found")
    return evaluation

@router.get("", response_model=list[EvaluationResponse])
async def list_evaluations(
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
    skip: int = 0,
    limit: int = 10
) -> list[EvaluationResponse]:
    """List user's evaluations"""
    service = EvaluationService(db)
    return await service.list(current_user.id, skip, limit)

@router.put("/{eval_id}", response_model=EvaluationResponse)
async def update_evaluation(
    eval_id: str,
    eval_data: EvaluationUpdate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
) -> EvaluationResponse:
    """Update evaluation"""
    service = EvaluationService(db)
    evaluation = await service.update(eval_id, current_user.id, eval_data)
    if not evaluation:
        raise HTTPException(status_code=404, detail="Evaluation not found")
    return evaluation

@router.delete("/{eval_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_evaluation(
    eval_id: str,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
):
    """Delete evaluation"""
    service = EvaluationService(db)
    success = await service.delete(eval_id, current_user.id)
    if not success:
        raise HTTPException(status_code=404, detail="Evaluation not found")

Streaming Route (Chat)

from fastapi.responses import StreamingResponse

@router.post("/api/chat/message")
async def send_message(
    message_data: ChatMessageCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
) -> StreamingResponse:
    """
    Send message to chat agent, stream response
    
    Streams back JSON lines: {"type": "content", "data": "text"}
    """
    service = ChatService(db)
    
    async def generate():
        async for chunk in service.stream_response(
            conversation_id=message_data.conversation_id,
            user_id=current_user.id,
            message=message_data.content
        ):
            # Send as JSON line
            yield f"data: {json.dumps(chunk)}\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

Admin Protected Route

from app.api.deps import get_admin_user

@router.post("/api/admin/documents/{doc_id}/pipeline")
async def start_pipeline(
    doc_id: str,
    config: PipelineConfig,
    db: AsyncSession = Depends(get_db),
    admin_user = Depends(get_admin_user)
):
    """Start document pipeline (admin only)"""
    service = DocumentPipelineService(db)
    return await service.start(doc_id, config)

Service Layer Pattern

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

class EvaluationService:
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def create(self, user_id: str, eval_data: EvaluationCreate):
        """Create evaluation with validation"""
        # Validate user exists
        user = await self.db.get(User, user_id)
        if not user:
            raise ValueError("User not found")
        
        # Create record
        evaluation = Evaluation(
            user_id=user_id,
            **eval_data.dict()
        )
        self.db.add(evaluation)
        await self.db.commit()
        await self.db.refresh(evaluation)
        return evaluation
    
    async def get(self, eval_id: str, user_id: str):
        """Get evaluation (owner only)"""
        stmt = select(Evaluation).where(
            (Evaluation.id == eval_id) &
            (Evaluation.user_id == user_id)
        )
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()
    
    async def list(self, user_id: str, skip: int = 0, limit: int = 10):
        """List user evaluations"""
        stmt = select(Evaluation)\
            .where(Evaluation.user_id == user_id)\
            .offset(skip)\
            .limit(limit)
        result = await self.db.execute(stmt)
        return result.scalars().all()

Auth & Dependencies

# app/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.database import get_session
from app.core.security import decode_access_token
from app.models.user import User

security = HTTPBearer()

async def get_db() -> AsyncSession:
    """Get async session"""
    async with get_session() as session:
        yield session

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db)
) -> User:
    """Get authenticated user"""
    token = credentials.credentials
    user_id = decode_access_token(token)
    
    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token invalide ou expiré"
        )
    
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

async def get_admin_user(
    current_user = Depends(get_current_user)
) -> User:
    """Get admin user (verify role)"""
    if current_user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return current_user

OpenAI Integration

from openai import AsyncOpenAI

class ChatService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
        self.rag_service = RAGService()
    
    async def stream_response(self, conversation_id: str, user_id: str, message: str):
        """Stream chat response with RAG context"""
        
        # Save user message
        user_msg = Message(
            conversation_id=conversation_id,
            role="user",
            content=message
        )
        self.db.add(user_msg)
        await self.db.commit()
        
        # Get RAG context
        context = await self.rag_service.search(message, limit=3)
        
        # Build prompt
        system_prompt = self._build_system_prompt(context)
        
        # Stream from OpenAI
        async with self.client.messages.stream(
            model="gpt-4-turbo-preview",
            max_tokens=1024,
            system=system_prompt,
            messages=[
                {"role": "user", "content": message}
            ]
        ) as stream:
            full_response = ""
            async for text in stream.text_stream:
                full_response += text
                yield {"type": "content", "data": text}
        
        # Save assistant response
        assistant_msg = Message(
            conversation_id=conversation_id,
            role="assistant",
            content=full_response,
            sources=context
        )
        self.db.add(assistant_msg)
        await self.db.commit()
        
        yield {"type": "done"}
    
    def _build_system_prompt(self, context: list) -> str:
        """Build system prompt with RAG context"""
        sources = "\n".join([f"- {c['title']}: {c['content'][:200]}..." for c in context])
        return f"""Tu es un expert ERP pour PME manufacturières.

Contexte de connaissances:
{sources}

Réponds aux questions de l'utilisateur en utilisant ce contexte.
Cite les sources quand tu utilises le contexte.
Sois concis et pratique."""

Error Handling

@router.post("/api/chat/message")
async def send_message(
    message_data: ChatMessageCreate,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user)
):
    """Send message with proper error handling"""
    try:
        service = ChatService(db)
        async for chunk in service.stream_response(...):
            yield chunk
    except ValueError as e:
        yield {"type": "error", "message": str(e)}
    except HTTPException as e:
        raise e
    except Exception as e:
        logger.error(f"Chat error: {e}")
        yield {"type": "error", "message": "Internal server error"}

Conventions

• Routes start with

  /api/

• Use descriptive HTTP methods (POST create, PUT update, DELETE delete)
• Always return status codes (201 create, 200 ok, 404 not found, 422 validation, 500 error)
• Docstring required on all route functions
• Use Pydantic schemas for request/response validation
• Service layer handles business logic
• Type hints everywhere
• Use

  async/await

  for async operations
• Error handling explicit with HTTPException
• Log important operations
• Pagination:

  skip

  and

  limit

  parameters