API RESTful: FAST API

Sitio:	Campus virtual DAW - damiansu
Curso:	Programación en Python
Libro:	API RESTful: FAST API

Imprimido por:	Invitado
Día:	domingo, 8 de marzo de 2026, 09:04

1. Explicación

2. FastAPI básico + Swagger

Proyecto: fastapi-demo

Abrimos el proyecto y activamos el entorno virtual

python -m venv venv

Windows

 .\venv\Scripts\Activate.ps1

Mac

source venv/bin/activate

Instalamos dependencias

pip install fastapi uvicorn

2.1. main.py

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(
    title="Demo FastAPI (sin BD)",
    description="Primero Swagger, luego MySQL",
    version="1.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)

@app.get("/")
def root():
    return {"ok": True, "mensaje": "FastAPI funcionando"}

@app.post("/productos")
def crear_producto(producto: ProductoCreate):
    # Simulación: devolvemos lo recibido + un id fijo
    return {"id": 1, **producto.model_dump()}

2.2. Arrancamos el server

uvicorn main:app --reload

Verificamos:

http://localhost:8000

http://localhost:8000/docs

Lanzamos un POST a /productos

Provocamos un 200 y un 422

3. Conectar con MySQL

3.1. Levantamos el servicio

Levantamos el servicio de MySQL

puerto: 3307
usuario: root
contraseña: ChuckNorris2025

Los datos a importar son estos;

CREATE DATABASE IF NOT EXISTS fastapi_db
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_unicode_ci;

USE fastapi_db;

DROP TABLE IF EXISTS productos;

CREATE TABLE productos (
    id INT AUTO_INCREMENT PRIMARY KEY,
    nombre VARCHAR(100) NOT NULL,
    precio DECIMAL(10,2) NOT NULL,
    stock INT NOT NULL DEFAULT 0
);

INSERT INTO productos (nombre, precio, stock) VALUES
('Portátil Lenovo', 899.99, 10),
('Monitor LG 27', 249.50, 15),
('Teclado Mecánico', 79.99, 30);

3.2. db.py

Instalamos dependencias

pip install sqlalchemy pymysql cryptography

Creamos db.py, cambiar valores:

puerto: 3307
usuario: root
contraseña: ChuckNorris2025

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "mysql+pymysql://root:ChuckNorris2025@localhost:3307/fastapi_db"

engine = create_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3.3. models.py

Creamos los modelos de las tablas, models.py

from sqlalchemy import Column, Integer, String, Numeric
from db import Base

class Producto(Base):
    __tablename__ = "productos"

    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(100), nullable=False)
    precio = Column(Numeric(10, 2), nullable=False)
    stock = Column(Integer, nullable=False, default=0)

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

3.5. Verificamos

Verificamos los métodos y comprobamos que el POST cambia el estado del base de datos:

Primeras conclusiones

Primero: API y documentación (Swagger)
Segundo: persistencia real (MySQL + ORM)
FastAPI no crea la BD: se integra con el ORM (como JPA en Java)

4. JWT BÁSICO EN FASTAPI

No metemos bases de datos en la autenciación

Partimos de:

FastAPI funcionando
Swagger visible (/docs)
Endpoint simple ya probado

4.1. Instalar dependencias JWT

Dependencias

bcrypt → contraseñas seguras
jose → firmar/verificar JWT
multipart → formulario de login en Swagger

pip install python-jose passlib[bcrypt<5] python-multipart

si no tira probad esta:

pip install python-jose passlib[bcrypt] python-multipart bcrypt<5

para mac

pip install python-jose "passlib[bcrypt]" python-multipart "bcrypt<5"

4.2. Crear auth.py

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from security import hash_password, verify_password, create_token

router = APIRouter(tags=["auth"])

# Usuario fijo SOLO para clase (sin BD)
FAKE_USER = {
    "username": "admin",
    "password_hash": hash_password("ChuckNorris2026")
}

@router.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends()):
    if form.username != FAKE_USER["username"]:
        raise HTTPException(status_code=401, detail="Usuario incorrecto")

    if not verify_password(form.password, FAKE_USER["password_hash"]):
        raise HTTPException(status_code=401, detail="Password incorrecto")

    token = create_token(form.username)
    return {"access_token": token, "token_type": "bearer"}

4.3. Crear security.py

from jose import jwt
from passlib.context import CryptContext

SECRET_KEY = "ChuckNorris2026"
ALGORITHM = "HS256"

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(password: str, password_hash: str) -> bool:
    return pwd_context.verify(password, password_hash)

def create_token(username: str) -> str:
    payload = {"sub": username}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> str:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    return payload["sub"]

4.4. Crear deps.py

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from security import decode_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        return decode_token(token)
    except JWTError:
        raise HTTPException(status_code=401, detail="Token inválido")

4.5. Modificar main.py

from fastapi import FastAPI, Depends
from auth import router as auth_router
from deps import get_current_user

app = FastAPI(
    title="FastAPI + Swagger + JWT (básico)",
    description="Login → token → endpoint protegido",
    version="1.0.0"
)

app.include_router(auth_router)

@app.get("/")
def root():
    return {"ok": True, "mensaje": "Ve a /docs para probar JWT"}

@app.get("/privado")
def privado(usuario: str = Depends(get_current_user)):
    return {"mensaje": f"Hola {usuario}, estás autenticado"}

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from auth import router as auth_router
from deps import get_current_user

from db import get_db
from models import Producto

app = FastAPI(
    title="FastAPI + Swagger + JWT (básico)",
    description="Login → token → endpoint protegido",
    version="1.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.include_router(auth_router)

@app.get("/")
def root():
    return {"ok": True, "mensaje": "API con MySQL lista. Ve a /docs"}

@app.get("/privado")
def privado(usuario: str = Depends(get_current_user)):
    return {"mensaje": f"Hola {usuario}, estás autenticado"}


@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

4.6. Verificamos con Swagger

4.7. Añadimos un endpoint con JWT

Explicación

/nombre -> el nombre lo saca del token

@app.get("/nombre")
def nombre(usuario: str = Depends(get_current_user)):
    return{"usuario": usuario}

4.8. Protegemos el POST

Explicación

@app.post("/productos", response_model=ProductoResponse, status_code=201)
def crear_producto(
    producto: ProductoCreate,
    db: Session = Depends(get_db),
    usuario: str = Depends(get_current_user)
):
    nuevo = Producto(
        nombre=producto.nombre,
        precio=producto.precio,
        stock=producto.stock
    )
    db.add(nuevo)
    db.commit()
    db.refresh(nuevo)
    return nuevo

4.9. PUT /productos/{producto_id} (PROTEGIDO)

Explicación

@app.put("/productos/{producto_id}", response_model=ProductoResponse)
def actualizar_producto(
    producto_id: int,
    producto: ProductoCreate,
    db: Session = Depends(get_db),
    usuario: str = Depends(get_current_user)
):
    existente = db.query(Producto).filter(Producto.id == producto_id).first()
    if not existente:
        raise HTTPException(status_code=404, detail="Producto no encontrado")

    existente.nombre = producto.nombre
    existente.precio = producto.precio
    existente.stock = producto.stock

    db.commit()
    db.refresh(existente)
    return existente

4.10. DELETE /productos/{producto_id} (PROTEGIDO)

@app.delete("/productos/{producto_id}", status_code=204)
def eliminar_producto(
    producto_id: int,
    db: Session = Depends(get_db),
    usuario: str = Depends(get_current_user)
):
    existente = db.query(Producto).filter(Producto.id == producto_id).first()
    if not existente:
        raise HTTPException(status_code=404, detail="Producto no encontrado")

    db.delete(existente)
    db.commit()
    return

API RESTful: FAST API

Tabla de contenidos

1. Explicación

2. FastAPI básico + Swagger

2.1. main.py

2.2. Arrancamos el server

3. Conectar con MySQL

3.1. Levantamos el servicio

3.2. db.py

3.3. models.py

3.4. main.py (modificación)

3.5. Verificamos

4. JWT BÁSICO EN FASTAPI

4.1. Instalar dependencias JWT

4.2. Crear auth.py

4.3. Crear security.py

4.4. Crear deps.py

4.5. Modificar main.py

4.6. Verificamos con Swagger

4.7. Añadimos un endpoint con JWT

4.8. Protegemos el POST

4.9. PUT /productos/{producto_id} (PROTEGIDO)

4.10. DELETE /productos/{producto_id} (PROTEGIDO)