Cómo Crear un CRUD con FastAPI para la Gestión de Usuarios y Login
Tabla de contenidos
- 1 Paso 1: Instalación de Dependencias
- 2 Paso 2: Estructura del Proyecto
- 3 Paso 3: Configuración del Archivo .env
- 4 Paso 4: Crear los Modelos de Base de Datos
- 5 Paso 5: Crear los Esquemas Pydantic
- 6 Paso 6: Configurar la Base de Datos
- 7 Paso 7: Crear Operaciones CRUD
- 8 Paso 8: Configurar Autenticación
- 9 Paso 9: Configurar Rutas de Usuario y Perfil
- 10 Paso 10: Configurar el Archivo main.py
- 11 Paso 11: Inicializar la Base de Datos
- 12 Paso 12: Ejecutar el Servidor
- 13 Paso 13: Configurar el Archivo .gitignore
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
- Crea un archivo
.env
en el directoriobackend
con los valores de configuración necesarios. - Crea un script
generate_secret.py
en el directoriobackend
para generar la clave secreta. - 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:
- Elimina las tablas existentes y crea nuevas tablas.
- Crea perfiles predeterminados para «admin» y «user».
- Crea usuarios predeterminados con perfiles «admin» y «user».
- 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.