postgres
databases
fast
Empezando por el boilerplate fastapi-hello, puedes encontrar un ejemplo de API funcionando con una base de datos. Todo el código de tu aplicación debe estar escrito dentro de la carpeta ./src/.
User.El uso de variables de entorno dentro del archivo .env ayuda a gestionar los ajustes de configuración en diferentes entornos (desarrollo, pruebas, producción) sin cambiar el código. FastAPI puede utilizar variables de entorno para la configuración a través de librerías como python-dotenv.
.env.example para crear un archivo .env. en la raíz de su proyecto:1DATABASE_URL=postgresql+psycopg2://gitpod:postgres@localhost:5432/example 2 3# Add any other variables below 4SOME_VAR=SOME_VALUE
El enrutamiento en FastAPI se realiza mediante el uso de objetos APIRouter que pueden ser incluidos en la aplicación principal.
./src/endpoints/user.py:1from fastapi import APIRouter, Depends 2from sqlalchemy.orm import Session 3from typing import List 4from .models import User as UserModel 5from .database import get_db 6from .utils import APIException 7from pydantic import BaseModel, EmailStr 8 9router = APIRouter() 10 11# Los serializadores se utilizan para validar el cuerpo de la solicitud entrante 12# Aquí se determina qué campos son obligatorios y sus tipos 13class CreateSerializer(BaseModel): 14 password: str = Field(..., min_length=3, max_length=50) 15 email: EmailStr 16 is_active: bool 17 18# Los serializadores también se utilizan para dar formato al cuerpo de la respuesta saliente 19class UserSmallSerializer(BaseModel): 20 email: str 21 is_active: bool 22 23 class Config: 24 from_attributes = True 25 26@router.get("/users/") 27def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): 28 users = db.query(UserModel).offset(skip).limit(limit).all() 29 return [UserSmallSerializer.model_validate(user) for user in users] 30 31@router.get("/users/{user_id}") 32def read_user(user_id: int, db: Session = Depends(get_db)): 33 user = db.query(UserModel).filter(UserModel.id == user_id).first() 34 if user is None: 35 raise APIException(status_code=404, detail="User not found") 36 return UserSmallSerializer.model_validate(user) 37 38@router.post("/users/") 39def create_user(user: CreateSerializer, db: Session = Depends(get_db)): 40 db_user = UserModel(username=user.username, email=user.email, age=user.age) 41 db.add(db_user) 42 db.commit() 43 db.refresh(db_user) 44 return UserSmallSerializer.model_validate(db_user) 45 46@router.put("/users/{user_id}", response_model=UserSmallSerializer) 47def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)): 48 db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 49 if db_user is None: 50 raise APIException(status_code=404, detail="User not found") 51 for key, value in user.dict().items(): 52 setattr(db_user, key, value) 53 db.commit() 54 db.refresh(db_user) 55 return UserSmallSerializer.model_validate(db_user) 56 57@router.delete("/users/{user_id}", response_model=UserSmallSerializer) 58def delete_user(user_id: int, db: Session = Depends(get_db)): 59 db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 60 if db_user is None: 61 raise APIException(status_code=404, detail="User not found") 62 db.delete(db_user) 63 db.commit() 64 return UserSmallSerializer.model_validate(db_user)
🔥 Todos los archivos python dentro de
./src/endpointsserán incluidos automáticamente como rutas en tu API, no hay necesidad de usar la funciónapp.include_router.
FastAPI proporciona validación automática de peticiones utilizando modelos Pydantic. Se pueden lanzar excepciones personalizadas para errores de validación.
El siguiente serializador llamado CreateSerializer fue definido para validar el POST /user que crea un nuevo usuario, el payload del endpoint debe contener una contraseña, email y el booleano is_active.
1# Los serializadores se utilizan para validar el cuerpo de la solicitud entrante 2# Here you determine which fields are required and their types 3class CreateSerializer(BaseModel): 4 password: str = Field(..., min_length=3, max_length=50) 5 email: EmailStr 6 is_active: bool
Tenemos que especificar nuestra clase CreateSerializer como primer parámetro de la función que maneja el método POST, en este caso la función create_user y FastAPI hará las validaciones:
1@router.post("/users/") 2# ⬇️ aquí añadimos el CreateSerializer 3def create_user(user: CreateSerializer, db: Session = Depends(get_db)): 4 db_user = UserModel(username=user.username, email=user.email, age=user.age) 5 db.add(db_user) 6 db.commit() 7 db.refresh(db_user) 8 return UserSmallSerializer.model_validate(db_user)
La serialización es manejada por modelos Pydantic que automáticamente convierten objetos Python a texto JSON.
./src/endpoints/user.py:
1@router.post("/users/", response_model=UserSmallSerializer) 2def create_user(user: UserCreate, db: Session = Depends(get_db)): 3 db_user = UserModel(username=user.username, email=user.email, age=user.age) 4 db.add(db_user) 5 db.commit() 6 db.refresh(db_user) 7 return UserSmallSerializer.model_validate(db_user)
Para las operaciones de base de datos utilizamos SQLAlchemy.
Define a database model in ./src/models.py:
1from sqlalchemy import Column, Integer, String 2from .database import Base 3 4class User(Base): 5 __tablename__ = "users" 6 id = Column(Integer, primary_key=True, index=True) 7 username = Column(String, unique=True, index=True) 8 email = Column(String, unique=True, index=True) 9 age = Column(Integer)
He aquí algunos ejemplos de creación, eliminación y actualización de un usuario:
1# Crear una nueva instancia de usuario SQLAlchemy 2db_user = UserModel(username=user.username, email=user.email, age=user.age) 3db.add(db_user) # Añadir el usuario a la sesión 4db.commit() # Confirmar la sesión para guardar el usuario en la base de datos 5 6# Eliminar un usuario por id 7db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 8if db_user is None: raise APIException(status_code=404, detail="User not found") 9db.delete(db_user) # Eliminar el usuario de la sesión 10db.commit() # Confirmar la sesión para eliminar el usuario de la base de datos 11 12# Actualizar el usuario por id 13db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 14if db_user is None: raise APIException(status_code=404, detail="User not found") 15# Actualizar los campos que necesite actualizar, por ejemplo: 16db_user.username = "some_new_username" 17# Confirmar la sesión para guardar los cambios en la base de datos 18db.commit() 19 20# Obtener todos los usuarios con más de 18 años de edad 21users = db.query(UserModel).filter(UserModel.age > 18).all()
Esta guía cubre la configuración básica y el uso de FastAPI con variables de entorno, enrutamiento (incluyendo operaciones CRUD), validaciones, serialización usando modelos Pydantic, y operaciones de base de datos con SQLAlchemy. Con estos componentes en su lugar, usted puede construir aplicaciones web escalables y robustas utilizando FastAPI.