Integración de bases de datos FastAPI
FastAPI se puede integrar con varias bases de datos; la forma más común es usar SQLAlchemy como ORM. En esta sección se explica cómo usar FastAPI + SQLAlchemy para construir una API con funcionalidades CRUD (Crear, Leer, Actualizar, Eliminar).
Instalación de dependencias
pip install sqlalchemy
Estructura del proyecto
project/ ├── main.py # FastAPI 应用入口 ├── database.py # 数据库连接配置 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 数据模型 └── crud.py # 数据库操作函数
1. Configuración de la base de datos
Endatabase.pyConfigurar la conexión de la base de datos en:
Ejemplo
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# Base de datos SQLite (adecuada para desarrollo)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Crear el motor de base de datos
# connect_args solo se necesita para SQLite, permite acceso multiproceso
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Crear la fábrica de sesiones
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Declarar la clase base
class Base(DeclarativeBase):
pass
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
# Base de datos SQLite (adecuada para desarrollo)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Crear el motor de base de datos
# connect_args solo se necesita para SQLite, permite acceso multiproceso
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# Crear la fábrica de sesiones
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Declarar la clase base
class Base(DeclarativeBase):
pass
Para el entorno de desarrollo es suficiente usar SQLite; es una base de datos de archivo que no requiere instalar servicios adicionales. Para producción se recomienda PostgreSQL o MySQL; solo hay que modificar
SQLALCHEMY_DATABASE_URLy listo.
2. Modelos SQLAlchemy
Enmodels.pyDefinir la estructura de las tablas de la base de datos en:
Ejemplo
# models.py
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Nombre de la tabla
id = Column(Integer, primary_key=True, index=True) # Clave primaria, autoincremental
email = Column(String, unique=True, index=True) # Único, crear índice
hashed_password = Column(String) # Contraseña hash
is_active = Column(Boolean, default=True) # Si está activo
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Título, crear índice
description = Column(String) # Descripción
owner_id = Column(Integer, ForeignKey("users.id")) # Clave foránea que relaciona con la tabla users
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users" # Nombre de la tabla
id = Column(Integer, primary_key=True, index=True) # Clave primaria, autoincremental
email = Column(String, unique=True, index=True) # Único, crear índice
hashed_password = Column(String) # Contraseña hash
is_active = Column(Boolean, default=True) # Si está activo
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Título, crear índice
description = Column(String) # Descripción
owner_id = Column(Integer, ForeignKey("users.id")) # Clave foránea que relaciona con la tabla users
3. Modelos de datos Pydantic
Enschemas.pyDefinir las estructuras de datos de entrada/salida de la API en:
Ejemplo
# schemas.py
from pydantic import BaseModel
# ===== Modelo de usuario =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Se requiere contraseña al crear
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Soporta objetos ORM
# ===== Modelo de producto =====
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class ItemOut(ItemBase):
id: int
owner_id: int
model_config = ConfigDict(from_attributes=True)
from pydantic import BaseModel
# ===== Modelo de usuario =====
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str # Se requiere contraseña al crear
class UserOut(UserBase):
id: int
is_active: bool
model_config = ConfigDict(from_attributes=True) # Soporta objetos ORM
# ===== Modelo de producto =====
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class ItemOut(ItemBase):
id: int
owner_id: int
model_config = ConfigDict(from_attributes=True)
Los modelos Pydantic y los modelos SQLAlchemy están separados. Pydantic se encarga de la validación de datos en la capa de API; SQLAlchemy se encarga de la estructura de tablas en la capa de base de datos.
from_attributes=TruePermite que Pydantic pueda leer datos desde objetos SQLAlchemy.
4. Funciones de operaciones de base de datos
Encrud.pyEncapsular las operaciones de base de datos en:
Ejemplo
# crud.py
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Obtener usuario por ID"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Obtener usuario por correo electrónico"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Obtener lista de usuarios"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Crear usuario"""
fake_hashed_password = user.password + "notreallyhashed" # En realidad se debería usar passlib para el hash
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Confirmar la transacción
db.refresh(db_user) # Refrescar el objeto para obtener el id generado por la base de datos
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Obtener lista de productos"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Crear producto"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
from sqlalchemy.orm import Session
import models, schemas
def get_user(db: Session, user_id: int):
"""Obtener usuario por ID"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
"""Obtener usuario por correo electrónico"""
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Obtener lista de usuarios"""
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
"""Crear usuario"""
fake_hashed_password = user.password + "notreallyhashed" # En realidad se debería usar passlib para el hash
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Confirmar la transacción
db.refresh(db_user) # Refrescar el objeto para obtener el id generado por la base de datos
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Obtener lista de productos"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Crear producto"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
5. Entrada principal de la aplicación FastAPI
Enmain.pyIntegrar todos los componentes en:
Ejemplo
# main.py
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict
import models, schemas, crud
from database import engine, SessionLocal, Base
# Crear tablas de la base de datos
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependencia: obtener sesión de base de datos
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== Rutas de usuario =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Comprobar si el correo ya existe
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=El correo ya está registrado)
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=El usuario no existe)
return db_user
# ===== Rutas de producto =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)
@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict
import models, schemas, crud
from database import engine, SessionLocal, Base
# Crear tablas de la base de datos
Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependencia: obtener sesión de base de datos
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# ===== Rutas de usuario =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Comprobar si el correo ya existe
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=El correo ya está registrado)
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=El usuario no existe)
return db_user
# ===== Rutas de producto =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)
@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items
Inyección de dependencias de la sesión de base de datos
El patrón principal es usar inyección de dependencias para gestionar las sesiones de base de datos:
def get_db():
db = SessionLocal() # 创建会话
try:
yield db # 提供会话给路由函数
finally:
db.close() # 请求完成后关闭会话
Cada solicitud obtiene una sesión de base de datos independiente, y al completarse la solicitud se cierra automáticamente para evitar fugas de recursos.
Cadenas de conexión para diferentes bases de datos
| Base de datos | Ejemplo de cadena de conexión |
|---|---|
| SQLite | sqlite:///./sql_app.db |
| PostgreSQL | postgresql://user:password@localhost/dbname |
| MySQL | mysql+pymysql://user:password@localhost/dbname |
Cambiar de base de datos solo requiere modificar la cadena de conexión; el resto del código no necesita cambios. Esta es una de las ventajas de usar ORM.
Resumen
- Usar SQLAlchemy como ORM; FastAPI gestiona las sesiones de base de datos mediante inyección de dependencias.
- Los modelos Pydantic (
schemas.py) se encargan de la validación de datos de la API, los modelos SQLAlchemy (models.py) se encargan de la estructura de tablas de la base de datos. - Las operaciones CRUD se encapsulan en
crud.py, manteniendo el código limpio. - En desarrollo usar SQLite; en producción cambiar a PostgreSQL/MySQL solo requiere modificar la cadena de conexión.
from_attributes=TruePermitir que Pydantic pueda leer datos de objetos ORM.