Cómo Crear un CRUD con FastAPI para la Gestión de Usuarios y Login

Paso 1: Instalación de Dependencias

Primero, crea un entorno virtual e instala las dependencias necesarias:

python3 -m venv env
source env/bin/activate

Crea el fichero requirements.txt con este contenido

fastapi
uvicorn[standard]
sqlalchemy
python-decouple
cryptography
python-jose[cryptography]
passlib[bcrypt]
bcrypt==3.2.0
pydantic

Este archivo requirements.txt asegurará que todas las dependencias necesarias estén instaladas cuando ejecutes el comando:

pip install -r requirements.txt

Este comando instalará todas las bibliotecas listadas en requirements.txt, configurando así tu entorno para ejecutar tu aplicación FastAPI.

Paso 2: Estructura del Proyecto

Asegúrate de tener la siguiente estructura de directorios y archivos:

fastCrudUsers/
├── backend/
│   ├── app/
│   │   ├── __init__.py
│   │   ├── main.py
│   │   ├── routers/
│   │   │   ├── __init__.py
│   │   │   ├── auth.py
│   │   │   ├── profile.py
│   │   │   ├── user.py
│   │   ├── schemas/
│   │   │   ├── __init__.py
│   │   │   ├── token.py
│   │   │   ├── user.py
│   │   │   ├── profile.py
│   │   ├── crud/
│   │   │   ├── __init__.py
│   │   │   ├── user.py
│   │   ├── db/
│   │   │   ├── __init__.py
│   │   │   ├── database.py
│   ├── .env
│   ├── create_project.sh
│   ├── generate_secret.py
│   ├── initialize_db.py
│   ├── requirements.txt
│   ├── test.db
├── fastapi-frontend/
└── .gitignore

Claro, vamos a agregar el script para crear la clave secreta y su ejecución al Paso 3. Este script generará una clave secreta y la agregará al archivo .env.

Paso 3: Configuración del Archivo .env

Crea un archivo .env en el directorio backend con el siguiente contenido, pero en lugar de definir manualmente la SECRET_KEY, usaremos un script para generarla.

.env

DATABASE_URL=sqlite:///./test.db
SECRET_KEY=
DEFAULT_PROFILE=user

Crear el Script para Generar la Clave Secreta

Crea un archivo generate_secret.py en el directorio backend para generar la clave secreta y agregarla al archivo .env.

generate_secret.py

from cryptography.fernet import Fernet
import os

# Generar una clave secreta
secret_key = Fernet.generate_key().decode()

# Especificar el archivo .env
env_file = ".env"

# Leer el contenido actual del archivo .env
if os.path.exists(env_file):
    with open(env_file, 'r') as f:
        lines = f.readlines()
else:
    lines = []

# Añadir o actualizar la clave secreta en el archivo .env
with open(env_file, 'w') as f:
    key_found = False
    for line in lines:
        if line.startswith("SECRET_KEY="):
            f.write(f'SECRET_KEY={secret_key}\n')
            key_found = True
        else:
            f.write(line)
    if not key_found:
        f.write(f'SECRET_KEY={secret_key}\n')

print(f"Clave secreta generada y guardada en {env_file}: {secret_key}")

Ejecutar el Script para Generar la Clave Secreta

Después de crear el archivo generate_secret.py, ejecuta el script para generar la clave secreta y actualizar el archivo .env.

cd backend
python generate_secret.py

Este comando generará una clave secreta y la agregará al archivo .env. El archivo .env se verá así después de ejecutar el script:

.env

DATABASE_URL=sqlite:///./test.db
SECRET_KEY=<clave_secreta_generada>
DEFAULT_PROFILE=user

Resumen del Paso 3

1. Crea un archivo .env en el directorio backend con los valores de configuración necesarios.
2. Crea un script generate_secret.py en el directorio backend para generar la clave secreta.
3. Ejecuta el script para generar la clave secreta y actualizar el archivo .env.

Este procedimiento asegura que tu SECRET_KEY sea única y segura, evitando la necesidad de definirla manualmente.

Paso 4: Crear los Modelos de Base de Datos

Define los modelos de base de datos en el directorio models.

app/models/user.py

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from app.db.database import Base

class User(Base):
    __tablename__ = 'users'

    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(200))
    is_active = Column(Integer, default=1)
    profile_id = Column(Integer, ForeignKey('profiles.id'))

    profile = relationship("Profile", back_populates="users")

app/models/profile.py

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.database import Base

class Profile(Base):
    __tablename__ = 'profiles'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, index=True)
    description = Column(String(200), index=True)

    users = relationship("User", back_populates="profile")

Paso 5: Crear los Esquemas Pydantic

Define los esquemas en el directorio schemas.

app/schemas/user.py

from pydantic import BaseModel
from typing import Optional

class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    password: str
    profile_id: Optional[int] = None

class User(UserBase):
    id: int
    is_active: bool
    profile_id: int

    class Config:
        from_attributes = True

app/schemas/profile.py

from pydantic import BaseModel
from typing import List, Optional

class ProfileBase(BaseModel):
    name: str
    description: Optional[str] = None

class ProfileCreate(ProfileBase):
    pass

class Profile(ProfileBase):
    id: int
    users: List["User"] = []

    class Config:
        from_attributes = True

app/schemas/token.py

from pydantic import BaseModel
from typing import Optional

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

Paso 6: Configurar la Base de Datos

Configura la conexión a la base de datos en database.py.

app/db/database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from decouple import config

DATABASE_URL = config('DATABASE_URL')

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

Paso 7: Crear Operaciones CRUD

Define las operaciones CRUD en crud.

app/crud/user.py

from sqlalchemy.orm import Session
from app.models import user as models
from app.schemas import user as schemas
from fastapi import HTTPException, status
from passlib.context import CryptContext
from app.models import profile as profile_models
from decouple import config

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

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

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

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    hashed_password = get_password_hash(user.password)
    profile_id = user.profile_id
    if not profile_id:
        default_profile = db.query(profile_models.Profile).filter(profile_models.Profile.name == DEFAULT_PROFILE).first()
        if not default_profile:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Default profile not found")
        profile_id = default_profile.id
    db_user = models.User(email=user.email, hashed_password=hashed_password, username=user.username, profile_id=profile_id)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def authenticate_user(db: Session, email: str, password: str):
    user = get_user_by_email(db, email)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

Paso 8: Configurar Autenticación

Define las rutas de autenticación en auth.py.

app/routers/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from jose import JWTError, jwt
from decouple import config
from app.schemas.token import Token, TokenData
from app.crud import user as crud_user
from app.db.database import get_db
from app.schemas.user import User

router = APIRouter()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = config("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = crud_user.get_user_by_email(db, email=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user

: User = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

async def get_current_admin_user(current_user: User = Depends(get_current_user)):
    if current_user.profile.name != "admin":
        raise HTTPException(status_code=403, detail="The user doesn't have enough privileges")
    return current_user

@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = crud_user.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

Paso 9: Configurar Rutas de Usuario y Perfil

Define las rutas de usuario y perfil en user.py y profile.py.

app/routers/user.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.crud import user as crud_user
from app.schemas import user as schemas_user
from app.db.database import get_db
from typing import List
from app.routers.auth import get_current_admin_user, get_current_active_user

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

@router.post("/", response_model=schemas_user.User, dependencies=[Depends(get_current_admin_user)])
def create_user(user: schemas_user.UserCreate, db: Session = Depends(get_db)):
    db_user = crud_user.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud_user.create_user(db=db, user=user)

@router.get("/", response_model=List[schemas_user.User], dependencies=[Depends(get_current_admin_user)])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud_user.get_users(db, skip=skip, limit=limit)
    return users

app/routers/profile.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.crud import profile as crud_profile
from app.schemas import profile as schemas_profile
from app.db.database import get_db
from typing import List
from sqlalchemy.exc import IntegrityError

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

@router.post("/", response_model=schemas_profile.Profile)
def create_profile(profile: schemas_profile.ProfileCreate, db: Session = Depends(get_db)):
    try:
        db_profile = crud_profile.create_profile(db=db, profile=profile)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Profile name already exists"
        )
    return db_profile

@router.get("/", response_model=List[schemas_profile.Profile])
def read_profiles(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    profiles = crud_profile.get_profiles(db, skip=skip, limit=limit)
    return profiles

Paso 10: Configurar el Archivo main.py

Define el archivo main.py para incluir las rutas.

app/main.py

from fastapi import FastAPI
from app.routers import user, profile, auth

app = FastAPI()

app.include_router(user.router)
app.include_router(profile.router)
app.include_router(auth.router)

@app.get("/")
def read_root():
    return {"message": "Welcome to fastCrud!"}

Paso 11: Inicializar la Base de Datos

Para crear un usuario para cada perfil (uno con perfil «admin» y otro con perfil «user»), puedes modificar el script initialize_db.py para agregar estos usuarios después de crear los perfiles. A continuación se muestra cómo puedes hacer esto:

initialize_db.py

from app.db.database import Base, engine
from app.models import user, profile
from sqlalchemy.orm import Session
from app.crud import user as crud_user, profile as crud_profile
from app.schemas import user as schemas_user, profile as schemas_profile
from decouple import config
import os

# Eliminar las tablas existentes y crear nuevas tablas
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)

# Crear una nueva sesión de base de datos
db = Session(bind=engine)

# Crear perfiles predeterminados
admin_profile = crud_profile.create_profile(db, schemas_profile.ProfileCreate(name="admin", description="Admin profile"))
user_profile = crud_profile.create_profile(db, schemas_profile.ProfileCreate(name="user", description="User profile"))

# Crear usuarios predeterminados
admin_user = schemas_user.UserCreate(
    username="admin",
    email="admin@example.com",
    password="adminpass",  # Cambia esto a una contraseña segura en producción
    profile_id=admin_profile.id
)
default_user = schemas_user.UserCreate(
    username="user",
    email="user@example.com",
    password="userpass",  # Cambia esto a una contraseña segura en producción
    profile_id=user_profile.id
)

# Guardar los usuarios en la base de datos
crud_user.create_user(db=db, user=admin_user)
crud_user.create_user(db=db, user=default_user)

# Confirmar las transacciones y cerrar la sesión
db.commit()
db.close()

print("Base de datos inicializada con perfiles y usuarios predeterminados.")

Este script:

1. Elimina las tablas existentes y crea nuevas tablas.
2. Crea perfiles predeterminados para «admin» y «user».
3. Crea usuarios predeterminados con perfiles «admin» y «user».
4. Guarda los usuarios en la base de datos.

Ejecutar el Script para Inicializar la Base de Datos

Después de crear o modificar el archivo initialize_db.py, ejecuta el script para inicializar la base de datos:

cd backend
python initialize_db.py

Este comando eliminará las tablas existentes, creará nuevas tablas, y agregará los perfiles y usuarios predeterminados a la base de datos. Al finalizar, deberías ver el mensaje:

Base de datos inicializada con perfiles y usuarios predeterminados.

Con esto, habrás completado la inicialización de tu base de datos con perfiles y usuarios predeterminados.

Paso 12: Ejecutar el Servidor

Asegúrate de estar en el directorio backend y de que el entorno virtual esté activado. Luego, ejecuta el servidor:

uvicorn app.main:app --reload

Paso 13: Configurar el Archivo .gitignore

Para asegurarte de que Git ignore los archivos y directorios que no deseas rastrear, necesitas configurar correctamente el archivo .gitignore. Aquí tienes un archivo .gitignore adecuado para tu proyecto, que incluye un backend de FastAPI y un frontend de JavaScript (React, Vue, etc.).

Crea un archivo .gitignore en la raíz de tu proyecto con el siguiente contenido:

.gitignore

# Entorno virtual de Python
env/
venv/
.venv/

# Archivos de configuración y entorno
.env
*.env

# Archivos de caché de Python
__pycache__/
*.py[cod]

# Archivos de bases de datos locales
*.db

# Archivos de logs
*.log

# Archivos temporales
*.tmp
*.temp
*.pid
*.swp
*.swo

# Directorios de configuración de IDEs
.idea/
.vscode/
*.sublime-project
*.sublime-workspace

# Directorios y archivos de dependencias de Node.js
node_modules/
npm-debug.log
yarn-error.log
yarn.lock
package-lock.json

# Archivos de compilación de frontend
dist/
build/

# Archivos de cobertura de pruebas
.coverage
coverage.xml
*.cover
*.py,cover
.hypothesis/

Este archivo .gitignore incluye las exclusiones comunes para un proyecto que utiliza Python para el backend y Node.js para el frontend. Asegúrate de incluir cualquier otro archivo o directorio específico de tu proyecto que no desees rastrear con Git.

¡Y eso es todo! Ahora tienes una aplicación FastAPI con un CRUD básico de gestión de usuarios, autenticación y perfiles de usuario.