API RESTful: FAST API
3. Conectar con MySQL
3.4. main.py (modificación)
Modificamos main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from db import get_db
from models import Producto
app = FastAPI(
title="FastAPI + Swagger + MySQL",
description="Sesión práctica: primero Swagger y luego persistencia real",
version="2.0.0"
)
class ProductoCreate(BaseModel):
nombre: str = Field(min_length=1, max_length=100)
precio: float = Field(gt=0)
stock: int = Field(ge=0, default=0)
class ProductoResponse(ProductoCreate):
id: int
class Config:
from_attributes = True # Pydantic v2
@app.get("/")
def root():
return {"ok": True, "mensaje": "API con MySQL lista. Ve a /docs"}
@app.get("/productos", response_model=list[ProductoResponse])
def listar_productos(db: Session = Depends(get_db)):
return db.query(Producto).all()
@app.get("/productos/{producto_id}", response_model=ProductoResponse)
def obtener_producto(producto_id: int, db: Session = Depends(get_db)):
prod = db.query(Producto).filter(Producto.id == producto_id).first()
if not prod:
raise HTTPException(status_code=404, detail="Producto no encontrado")
return prod
@app.post("/productos", response_model=ProductoResponse, status_code=201)
def crear_producto(producto: ProductoCreate, db: Session = Depends(get_db)):
nuevo = Producto(nombre=producto.nombre, precio=producto.precio, stock=producto.stock)
db.add(nuevo)
db.commit()
db.refresh(nuevo)
return nuevo