Manual Completo de FastAPI con PostgreSQL Schemas Separados

Implementación de un sistema de inventario con autenticación JWT y esquemas PostgreSQL separados

🚀 Introducción

Este manual cubre el desarrollo completo de una API RESTful para gestión de inventario utilizando FastAPI con PostgreSQL, implementando:

💡 Características Principales

📐 Arquitectura del Proyecto

app/
├── main.py                  # Punto de entrada
├── database.py              # Configuración DB
├── models/                  # Modelos SQLAlchemy
│   ├── auth_models.py       # Modelos para autenticación (schema auth)
│   └── inventory_models.py  # Modelos para inventario (schema inventory)
├── schemas/                 # Esquemas Pydantic
│   ├── auth.py
│   └── inventory.py
├── routers/                 # Routers FastAPI
│   ├── auth.py
│   └── inventory.py
├── dependencies.py          # Dependencias compartidas
├── utils/                   # Utilidades
│   └── security.py          # JWT y BCrypt
├── requirements.txt         # Dependencias Python
└── .env                     # Variables de entorno

1. Configuración del Proyecto

Instalación de Dependencias

Terminal
# Crear entorno virtual
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# Instalar dependencias
pip install fastapi uvicorn sqlalchemy psycopg2-binary python-dotenv python-jose[cryptography] passlib[bcrypt]

Estructura de PostgreSQL con Schemas Separados

SQL
-- Crear schemas separados
CREATE SCHEMA auth;
CREATE SCHEMA inventory;

-- Tabla de usuarios en schema auth
CREATE TABLE auth.users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    hashed_password TEXT NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Tabla de productos en schema inventory
CREATE TABLE inventory.products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    description TEXT,
    price NUMERIC(10, 2) NOT NULL,
    stock INTEGER DEFAULT 0,
    category VARCHAR(50),
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

2. Configuración de PostgreSQL

Configuración de SQLAlchemy

app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv
import os

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")

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

Base = declarative_base()

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

Modelos SQLAlchemy con Schemas

app/models/auth_models.py
from sqlalchemy import Column, Integer, String, Boolean, TIMESTAMP
from sqlalchemy.sql import func
from ..database import Base

class User(Base):
    __tablename__ = "users"
    __table_args__ = {"schema": "auth"}
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(TIMESTAMP, server_default=func.now())
app/models/inventory_models.py
from sqlalchemy import Column, Integer, String, Numeric, TIMESTAMP
from sqlalchemy.sql import func
from ..database import Base

class Product(Base):
    __tablename__ = "products"
    __table_args__ = {"schema": "inventory"}
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), index=True)
    description = Column(String)
    price = Column(Numeric(10, 2))
    stock = Column(Integer, default=0)
    category = Column(String(50))
    created_at = Column(TIMESTAMP, server_default=func.now())
    updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now())

🔍 Variables de Entorno (.env)

DATABASE_URL=postgresql://user:password@localhost:5432/inventory_db
SECRET_KEY=tu_super_secreto_jwt
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

3. Autenticación Segura

Utilidades de Seguridad

app/utils/security.py
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
from dotenv import load_dotenv
import os

load_dotenv()

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

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES"))

def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str):
    return pwd_context.hash(password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

Router de Autenticación

app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from ..schemas.auth import Token
from ..models.auth_models import User
from ..database import get_db
from ..utils.security import (
    verify_password,
    create_access_token,
    get_password_hash
)

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

@router.post("/token", response_model=Token)
def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=400,
            detail="Incorrect username or password"
        )
    
    access_token = create_access_token(
        data={"sub": user.username}
    )
    return {"access_token": access_token, "token_type": "bearer"}

@router.post("/register")
def register_user(username: str, email: str, password: str, db: Session = Depends(get_db)):
    hashed_password = get_password_hash(password)
    db_user = User(
        username=username,
        email=email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return {"message": "User created successfully"}

4. CRUD de Inventario

Esquemas Pydantic

app/schemas/inventory.py
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class ProductBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    stock: int = 0
    category: Optional[str] = None

class ProductCreate(ProductBase):
    pass

class Product(ProductBase):
    id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        orm_mode = True

Router de Productos

app/routers/inventory.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from ..schemas.inventory import Product, ProductCreate
from ..models.inventory_models import Product as DBProduct
from ..database import get_db
from ..utils.security import get_current_user

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

@router.post("/", response_model=Product)
def create_product(
    product: ProductCreate,
    db: Session = Depends(get_db),
    current_user: str = Depends(get_current_user)
):
    db_product = DBProduct(**product.dict())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

@router.get("/", response_model=List[Product])
def read_products(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    products = db.query(DBProduct).offset(skip).limit(limit).all()
    return products

@router.get("/{product_id}", response_model=Product)
def read_product(product_id: int, db: Session = Depends(get_db)):
    product = db.query(DBProduct).filter(DBProduct.id == product_id).first()
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@router.put("/{product_id}", response_model=Product)
def update_product(
    product_id: int,
    product: ProductCreate,
    db: Session = Depends(get_db),
    current_user: str = Depends(get_current_user)
):
    db_product = db.query(DBProduct).filter(DBProduct.id == product_id).first()
    if db_product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    
    for key, value in product.dict().items():
        setattr(db_product, key, value)
    
    db.commit()
    db.refresh(db_product)
    return db_product

@router.delete("/{product_id}")
def delete_product(
    product_id: int,
    db: Session = Depends(get_db),
    current_user: str = Depends(get_current_user)
):
    product = db.query(DBProduct).filter(DBProduct.id == product_id).first()
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    
    db.delete(product)
    db.commit()
    return {"message": "Product deleted successfully"}

5. Configuración Completa

Punto de Entrada FastAPI

app/main.py
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from .routers import auth, inventory
from .database import engine, Base
from .models import auth_models, inventory_models

# Crear tablas en la base de datos
Base.metadata.create_all(bind=engine)

app = FastAPI()

# Configurar CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Incluir routers
app.include_router(auth.router)
app.include_router(inventory.router)

@app.get("/")
def read_root():
    return {"message": "Inventory API with FastAPI"}

Dependencias

requirements.txt
fastapi==0.95.2
uvicorn==0.22.0
sqlalchemy==2.0.15
psycopg2-binary==2.9.6
python-dotenv==1.0.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4

Ejecutar la Aplicación

Terminal
uvicorn app.main:app --reload

6. Documentación con Swagger UI

📚 Documentación Interactiva

FastAPI genera automáticamente documentación interactiva:

GET http://localhost:8000/docs (Swagger UI)

GET http://localhost:8000/redoc (ReDoc)

Swagger UI

Ejemplo de Autenticación en Swagger

  1. Accede a /docs
  2. Busca el endpoint /token
  3. Haz clic en "Try it out"
  4. Ingresa:
    • username: tu_usuario
    • password: tu_contraseña
  5. Ejecuta y copia el token devuelto
  6. Haz clic en el botón "Authorize" (arriba a la derecha)
  7. Ingresa: Bearer TU_TOKEN_AQUI
  8. Ahora puedes probar los endpoints protegidos