Construyendo una API CRUD completa
En esta leccion, construiremos una API REST CRUD (Crear, Leer, Actualizar, Eliminar) completa usando FastAPI con SQLAlchemy para acceso a bases de datos. Este es un patron listo para produccion usado en aplicaciones del mundo real. Cubriremos enrutamiento, modelos de base de datos, esquemas, manejo de errores y codigos de respuesta.
Estructura del proyecto
# Recommended project structure:
# app/
# __init__.py
# main.py # FastAPI app and routes
# models.py # SQLAlchemy ORM models
# schemas.py # Pydantic schemas
# database.py # Database connection
# crud.py # Database operations
Configuracion de la base de datos
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
DATABASE_URL = "sqlite:///./app.db"
# For PostgreSQL: "postgresql://user:pass@localhost/dbname"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Modelos SQLAlchemy
# models.py
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime, timezone
from .database import Base
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False, index=True)
description = Column(String(500), nullable=True)
price = Column(Float, nullable=False)
in_stock = Column(Boolean, default=True)
category = Column(String(50), nullable=False)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc))
reviews = relationship("Review", back_populates="product", cascade="all, delete-orphan")
class Review(Base):
__tablename__ = "reviews"
id = Column(Integer, primary_key=True, index=True)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
rating = Column(Integer, nullable=False)
comment = Column(String(500), nullable=True)
author = Column(String(100), nullable=False)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
product = relationship("Product", back_populates="reviews")
Esquemas Pydantic
# schemas.py
from pydantic import BaseModel, Field
from datetime import datetime
class ProductCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
description: str | None = Field(None, max_length=500)
price: float = Field(..., gt=0)
category: str = Field(..., min_length=1, max_length=50)
in_stock: bool = True
class ProductUpdate(BaseModel):
name: str | None = Field(None, min_length=1, max_length=100)
description: str | None = Field(None, max_length=500)
price: float | None = Field(None, gt=0)
category: str | None = Field(None, min_length=1, max_length=50)
in_stock: bool | None = None
class ProductResponse(BaseModel):
id: int
name: str
description: str | None
price: float
category: str
in_stock: bool
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class ProductList(BaseModel):
items: list[ProductResponse]
total: int
skip: int
limit: int
Operaciones CRUD
# crud.py
from sqlalchemy.orm import Session
from sqlalchemy import func
from . import models, schemas
def get_products(
db: Session, skip: int = 0, limit: int = 10,
category: str | None = None, in_stock: bool | None = None
) -> tuple[list[models.Product], int]:
query = db.query(models.Product)
if category:
query = query.filter(models.Product.category == category)
if in_stock is not None:
query = query.filter(models.Product.in_stock == in_stock)
total = query.count()
products = query.offset(skip).limit(limit).all()
return products, total
def get_product(db: Session, product_id: int) -> models.Product | None:
return db.query(models.Product).filter(models.Product.id == product_id).first()
def create_product(db: Session, product: schemas.ProductCreate) -> models.Product:
db_product = models.Product(**product.model_dump())
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
def update_product(
db: Session, product_id: int, product: schemas.ProductUpdate
) -> models.Product | None:
db_product = get_product(db, product_id)
if not db_product:
return None
update_data = product.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(db_product, key, value)
db.commit()
db.refresh(db_product)
return db_product
def delete_product(db: Session, product_id: int) -> bool:
db_product = get_product(db, product_id)
if not db_product:
return False
db.delete(db_product)
db.commit()
return True
Rutas de la API
# main.py
from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from . import crud, schemas, models
from .database import engine, get_db
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="Product API", version="1.0.0")
@app.post("/products", response_model=schemas.ProductResponse, status_code=201)
def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)):
return crud.create_product(db, product)
@app.get("/products", response_model=schemas.ProductList)
def list_products(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
category: str | None = None,
in_stock: bool | None = None,
db: Session = Depends(get_db),
):
products, total = crud.get_products(db, skip, limit, category, in_stock)
return {"items": products, "total": total, "skip": skip, "limit": limit}
@app.get("/products/{product_id}", response_model=schemas.ProductResponse)
def get_product(product_id: int, db: Session = Depends(get_db)):
product = crud.get_product(db, product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.patch("/products/{product_id}", response_model=schemas.ProductResponse)
def update_product(
product_id: int, product: schemas.ProductUpdate, db: Session = Depends(get_db)
):
updated = crud.update_product(db, product_id, product)
if not updated:
raise HTTPException(status_code=404, detail="Product not found")
return updated
@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, db: Session = Depends(get_db)):
if not crud.delete_product(db, product_id):
raise HTTPException(status_code=404, detail="Product not found")
Puntos clave
- Separa responsabilidades: Modelos, esquemas, operaciones CRUD y rutas en archivos separados
- Usa Pydantic para validacion: Esquemas separados para crear/actualizar/respuesta
- Codigos HTTP adecuados: 201 para crear, 204 para eliminar, 404 para no encontrado
- Inyeccion de dependencias: Usa Depends(get_db) para sesiones de base de datos