User: I need a FastAPI endpoint that receives a CSV file, validates each row with Pydantic, and returns a summary of valid vs invalid rows. --- Here's a complete, production-ready implementation: from fastapi import FastAPI, UploadFile, File, HTTPException from pydantic import BaseModel, EmailStr, validator import csv, io from typing import List app = FastAPI() class UserRow(BaseModel): name: str email: EmailStr age: int @validator('age') def age_must_be_positive(cls, v): if v < 0 or v > 120: raise ValueError('Age must be between 0 and 120') return v class ProcessingResult(BaseModel): total_rows: int valid_rows: int invalid_rows: int errors: List[dict] @app.post("/upload-users", response_model=ProcessingResult) async def upload_users(file: UploadFile = File(...)): if not file.filename.endswith('.csv'): raise HTTPException(status_code=400, detail="Only CSV files accepted") content = await file.read() decoded = content.decode('utf-8') reader = csv.DictReader(io.StringIO(decoded)) valid, invalid, errors = 0, 0, [] for i, row in enumerate(reader, start=2): # start=2 because row 1 is header try: UserRow(**row) valid += 1 except Exception as e: invalid += 1 errors.append({"row": i, "data": row, "error": str(e)}) return ProcessingResult( total_rows=valid + invalid, valid_rows=valid, invalid_rows=invalid, errors=errors ) # Key decisions: # - UploadFile streams the file instead of loading it all in memory # - Pydantic validator runs field-level validation automatically # - Errors include row number so user can fix the source CSV # - response_model ensures the output schema is always consistent