Cómo Crear una Aplicación CRUD con FastAPI

Índice

Introducción
Estructura de Archivos y Carpetas
Instalación y Configuración
Conceptos Fundamentales
Flujo de Trabajo en FastAPI
Comandos Esenciales
Ejemplo Completo: Aplicación CRUD
Probando las Rutas con la Documentación de FastAPI
Despliegue en Producción
Recursos Adicionales

1. Introducción

FastAPI es un moderno y rápido (de alto rendimiento) framework web para construir APIs con Python 3.6+ basado en estándares Python. En este README, aprenderemos cómo crear una aplicación CRUD (Create, Read, Update, Delete) utilizando FastAPI.

🚀 FastAPI permite crear APIs RESTful de manera rápida y eficiente, con validación automática, serialización y documentación interactiva.

💻 Desarrollado por Sebastián Ramírez, FastAPI se basa en Starlette para el manejo web y Pydantic para la validación de datos, lo que lo hace extremadamente rápido y fácil de usar.

🔧 FastAPI es ideal para microservicios, aplicaciones serverless, y APIs que requieren alto rendimiento y facilidad de desarrollo.

2. Estructura de Archivos y Carpetas

book_crud/
│
├── main.py                   
├── config
│   ├── __init__.py
│   └── config_variables.py
|
├── database
│   ├── __init__.py
│   └── database.py
|                   
├── models/
│   ├── __init__.py
│   └── libro_model.py
|
├── schemas
│   ├── __init__.py
│   └── libro_schema.py
|
├── routes
│   ├── __init__.py
│   └── routes.py         
│
├── controllers/
│   ├── __init__.py
│   └── libro_controller.py   
│
├── .env #opcional
│
└── db.sqlite3 #Este archivo se creará solo ;)

venv/: Directorio del entorno virtual de Python.
__init__.py: Archivo que convierte el directorio en un paquete Python.
main.py: Contiene la instancia principal de la aplicación FastAPI y laconfiguración global.
config_variables.py: Guarda las variables de entorno utilizando pydantic_settings.
database.py: Maneja la configuración y conexión a la base de datos.
models.py: Define los modelos de SQLAlchemy que representan las tablas de la base de datos.
schemas.py: Contiene los esquemas Pydantic para la validación de datos yserialización.
controllers.py: Implementa la lógica de negocio y las operaciones CRUD.
routes.py: Define las rutas y endpoints de la API.
requirements.txt: Lista todas las dependencias del proyecto para una fácil instalación.
README.md: Proporciona documentación e instrucciones para el proyecto (este archivo).

3. Instalación y Configuración

Crear la estructura de directorios:

mkdir book_crud
cd book_crud

Instalar FastAPI y dependencias

Crear un entorno virtual:

python -m venv venv
source venv/bin/activate

Instalar FastAPI y dependencias:

pip install fastapi[all] sqlalchemy

si no te permite utilizar [all] entonces instala:

pip install fastapi sqlalchemy uvicorn pymysql

Crea las carpetas que necesites y dentro los archivos que vayasa utilizar, la arquitectura de tu proyecto puede cambiar, sinembargo recuerda que queremos escalabilidad, por lo quenecesitamos dividir la lógica de los distintos servicios, y laconexiones con otras partes de la aplicación, es decir crea lascarpetas y archivos (los archivos son los que tienen extensionescomo ".py", las carpetas no tienen extensión):

book_crud/
│
├── main.py                   
├── config
│  ├── __init__.py
│  └── config_variables.py
|
├── database
│   ├── __init__.py
│   └── database.py
|                   
├── models/
│   ├── __init__.py
│   └── libro_model.py
|
├── schemas
│   ├── __init__.py
│   └── libro_schema.py
|
├── routes
│   ├── __init__.py
│   └── routes.py         
│
├── controllers/
│   ├── __init__.py
│   └── libro_controller.py   
│ 
└──.env #opcional

Configurar las variables de entorno y usar el diectorio y archivo config/config_variables.py:

Para eso primero debemos instalar la dependencia pertinente, en este caso pydantic_settings, librería que nos da una serie de herramientas para permitir que nuestro sistema pueda acceder a información pertinente durante el desarrollo.

Pydantic es una librería de Python que se utiliza para validar, convertir y estructurar datos usando tipado de Python. Pydantic se asegura de que los datos que entran y salen de tu aplicación tengan la forma, el tipo y el contenido correcto.
```
pip install pydantic-settings
```
Nos dirigimos al directorio config/ y dentro de esta carpeto nos dirigimos al archivo config_variables.py y escribimos:
```
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DB_USER: str = "nombre-del-ddbb-user"
    DB_PASSWORD: str = "contraseña-ddbb"
    DB_HOST: str = "ddbb-host"
    DB_NAME: str = "nombre-ddbb"

settings = Settings()
```
Pero si quisieramos hacer nuestro entorno más seguro podriamos usar el paquete python-dotenv y un archivo .env:
```
pip install python-dotenv
```
Ahora para guardar estas dependencias con sus versiones en un archivo requirements.txt hacemos:
```
pip freeze >> requirements.txt
```
Luego en tu archivo .env:
```
DB_USER=nombre-del-ddbb-user
DB_PASSWORD=contraseña-ddbb
DB_HOST=ddbb-host
DB_NAME=nombre-ddbb
```
Entonces el archivo config/config_variables.py se veria de esta manera:
```
 # config/config_variables.py

 import os
 from dotenv import load_dotenv
   
 # Cargar variables del .env
 load_dotenv()
   
 class Settings:
     DB_USER: str = os.getenv("DB_USER", "default_user")
     DB_PASSWORD: str = os.getenv("DB_PASSWORD","default_password")
     DB_HOST: str = os.getenv("DB_HOST", "localhost")
     DB_NAME: str = os.getenv("DB_NAME", "test_db")
   
 settings = Settings()
```

Configurar la base de datos en database/database.py y en MySQL Workbench:

# database/database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from config.config_variables import settings

 # Variables de entorno para no exponer información sensible
DB_USER = settings.DB_USER
DB_PASSWORD = settings.DB_PASSWORD
DB_HOST = settings.DB_HOST
DB_NAME = settings.DB_NAME

# Conexión con la base de datos
DATABASE_URL = "mysql+pymysql://"+DB_USER+":"+DB_PASSWORD+"@"+DB_HOST+"/"+DB_NAME+""  # MySQL

# Crea un engine
engine = create_engine(DATABASE_URL)

# Crea una clase para configurar la sesión
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Crea una clase base para los modelos
Base = declarative_base()

# función para obtener la sesión de la base de datos
def get_db():
    db = Session()  # Crea una nueva sesión
    try:
        yield db  # Usa la sesión
    finally:
        db.close()  # Cierra la sesión al terminar

# Esta función crea una sesión para trabajar con la base de datos, la devuelve mientras haces algo (yield db) y la cierra automáticamente al terminar.
# Se usa mucho en frameworks como FastAPI para que cada petición tenga su propia sesión limpia.

Creamos nuestra conexión en Workbech

Ahora entramos a la conexión y creamos una base de datos

Nos aseguramos de estar en schemas y no en Administration

Creamos la base de datos

Le ponemos como nombre book_crud y hacemos click en el botón apply que está al final de esa vista:

Si no ves tu base de datos, o que haya habido alg[un cambio, refresca haciendo click en estas flechas:

Hacemos click derecho sobre la opción tables (la cual está dentro de book_crud en el menú lateral), y hacemos click en create table para así crear una nueva tabla:

A la tabla le colocaremos como nombre libros y sus atributos serán id, title, description y volvemos a hacer click en apply:

Comprobamos que estamos correctamente conectados a la base de datos

Para comprobar que estamos conectados a la base de datos necesitamos que nuestro archivo main.py tenga definido el arranque de la aplicación:

from fastapi import FastAPI

app = FastAPI()

#IMPORTANTE: Aquí referenciamos el archivo "database" no la variable "db"
from database import database

def run():
    pass
if __name__ == '__main__':
    database.Base.metadata.create_all(database.engine)
    run()

@app.get("/")
async def root():
    return {"message": "Hello World"}

Y luego en nuestra terminal ejecutamos:

uvicorn main:app --reload

4. Conceptos Fundamentales

create_engine: Piensa en esto como construir la tubería que conecta tu código con la base de datos.
declarative_base: Es una plantilla para crear tus tablas como si fueran clases de Python.
sessionmaker: Es como un generador de “mesas de trabajo” para interactuar con la base de datos sin tocarla directamente. Con la sesión puedes consultar, insertar, actualizar o borrar datos sin afectar inmediatamente la base de datos real hasta que confirmes los cambios.
- autocommit=False → No confirma automáticamente los cambios, tú decides cuándo guardar.
- autoflush=False → No envía automáticamente los cambios hasta que decidas.
- bind=engine → Le decimos a la sesión que use esa tubería que creamos antes para conectarse.
Modelos: Representan las tablas en la base de datos.
Schemas: Definen la estructura de los datos para la serialización/deserialización.
CRUD: Operaciones básicas (Create, Read, Update, Delete) para manipular datos.
Dependencias: Funciones que FastAPI ejecuta antes de las funciones de ruta.
Pydantic: Biblioteca para validación de datos y configuraciones.

5. Flujo de Trabajo en FastAPI

Definir modelos de base de datos
Crear schemas Pydantic
Implementar operaciones CRUD
Definir rutas de la API
Configurar la aplicación principal

6. Comandos Esenciales

fastapi run main.py: Inicia el servidor de desarrollo
uvicorn main:app --reload: Inicia el servidor de desarrollo
pytest: Ejecuta las pruebas (si están configuradas)
alembic revision --autogenerate: Genera una migración de base de datos (si se usa Alembic)
alembic upgrade head: Aplica las migraciones pendientes

7. Ejemplo Completo: Aplicación CRUD

Modelos (`models/libro_model.py`)

from sqlalchemy import Column, Integer, String
from database.database import Base

class Libro(Base):
    __tablename__ = "libros"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True, nullable=False)
    description = Column(String, index=True)

Schemas (`schemas/libro_schema.py`)

from pydantic import BaseModel

# Hereda todo de ItemBase.
# Se usa cuando el usuario envía datos para crear un item.
# “pass” significa que no añadimos nada nuevo, solo usamos lo que está en ItemBase.

class LibroBase(BaseModel):
    title: str
    description: str = None

class LibroCreate(LibroBase):
    pass

class Libro(LibroBase): # Hereda de ItemBase (title y description) y añade el id que ya existe en la base de datos.
    id: int

    class Config:
        orm_mode = True # permite que Pydantic lea directamente objetos de SQLAlchemy como si fueran diccionarios, para enviarlos en respuestas JSON.

Operaciones CRUD (`controllers/libro_controller.py`)

from sqlalchemy.orm import Session
from schemas import libro_schema
from models.libro_model import Libro

class CrudControllers:

    '''
    Si no usaramos @static methos tendriamos que escribir:

    def __init__(self, db: Session): # El trabajador ya tiene su caja de herramientas (self.db) y no necesita traer nada externo cada vez.
            self.db = db

    y el controlador se vería así:

    def create_item(self, item: crud_schema.ItemCreate): # No hace falta pasar db:session cada vez
            db_item = Item(**item.dict())
            self.db.add(db_item)
            self.db.commit()
            self.db.refresh(db_item)
            return db_item
    '''

    @staticmethod
    def get_Libros(db: Session):
        return db.query(Libro).all()

    @staticmethod # Cada función es como un trabajador independiente que llega con su propio conjunto de herramientas (db) y hace su tarea.
    def get_Libro_by_id(db: Session, Libro_id: int):
        return db.query(Libro).filter(Libro.id == Libro_id).first()

    @staticmethod
    def create_Libro(db: Session, libro: libro_schema.LibroCreate):
        db_libro = Libro(**libro.dict())  # libro.dict() funciona porque es un Pydantic model
        db.add(db_libro)
        db.commit()
        db.refresh(db_libro)
        return db_libro

    @staticmethod
    def update_Libro(db: Session, Libro_id: int, libro: libro_schema.LibroCreate):
        db_Libro = db.query(Libro).filter(Libro.id == Libro_id).first()
        if db_Libro:
            db_Libro.title = libro.title 
            db_Libro.description = libro.description
            db.commit()
            db.refresh(db_Libro)
        return db_Libro
   
    @staticmethod
    def delete_Libro(db: Session, Libro_id: int):
        db_Libro = db.query(Libro).filter(Libro.id == Libro_id).first()
        if db_Libro:
            db.delete(db_Libro)
            db.commit()
        return {"message": "Libro eliminado exitosamente"}

También nuestros controladores pueden ser funciones independientes, no tienen por qué estar dentro de una clase:

from sqlalchemy.orm import Session
from schemas import libro_schema
from models.libro_model import Libro

def get_Libros(db: Session):
    return db.query(Libro).all()

def get_Libro_by_id(db: Session,Libro_id: int):
    return db.query(Libro).filter(Libro.id == Libro_id).first()


def create_Libro(db: Session, libro:libro_schema.LibroCreate):
    db_libro = Libro(**libro.dict()) # libro.dict() funciona porque esun Pydantic model
    db.add(db_libro)
    db.commit()
    db.refresh(db_libro)
    return db_libro


def update_Libro(db: Session,Libro_id: int, libro: libro_schemaLibroCreate):
    db_Libro = db.query(Libro).filte(Libro.id == Libro_id).first()
    if db_Libro:
        db_Libro.title = libro.title 
        db_Libro.description = librodescription
        db.commit()
        db.refresh(db_Libro)
    return db_Libro
   

def delete_Libro(db: Session,Libro_id: int):
    db_Libro = db.query(Libro).filte(Libro.id == Libro_id).first()
    if db_Libro:
        db.delete(db_Libro)
        db.commit()
    return {"message": "Libroeliminado exitosamente"}

Rutas de la API (`routes/routes.py`)

from fastapi import APIRouter, Depends
from fastapi import status
from sqlalchemy.orm import Session
from typing import List

from controllers.crud_controller import CrudController
from schemas.attendance_schema import AttendanceSchema
from database.database import get_db

router = APIRouter()

@router.post("/items/", response_model=schemas.Item)
def new_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    item = await CrudController.create_item(item, db)
    return item

@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = CrudController.get_item(db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# Obtener lista de items
@router.get("/items/", response_model=List[crud_schema.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = CrudControllers.get_items(db, skip=skip, limit=limit)
    return items

# Actualizar un item
@router.put("/items/{item_id}", response_model=crud_schema.Item)
def update_item(item_id: int, item: crud_schema.ItemCreate, db: Session = Depends(get_db)):
    db_item = CrudControllers.update_item(db, item_id, item)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# Borrar un item
@router.delete("/items/{item_id}", response_model=crud_schema.Item)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = CrudControllers.delete_item(db, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

Si definimos los controladores como funciones en lugar de una clase, entonces nuestras rutas se van a ver de esta manera:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from controllers.libro_controller import CrudControllers
from schemas.libro_schema import LibroBase, LibroCreate, Libro
from database.database import get_db

router = APIRouter()

@router.post("/libros/", response_model=LibroBase, status_code=status.HTTP_201_CREATED)
async def new_libro(libro: LibroCreate, db: Session = Depends(get_db)):
    libro = CrudControllers.create_Libro(db,libro)
    return libro

@router.get("/libros/{libro_id}", response_model=LibroBase)
def get_libro(libro_id: int, db: Session = Depends(get_db)):
    libro = CrudControllers.get_Libro_by_id(db, libro_id)
    if not libro:
        raise HTTPException(status_code=404, detail="Libro no encontrado")
    return libro

@router.put("/libros/{libro_id}", response_model=LibroBase)
def update_Libro(libro_id: int, libro: LibroCreate, db: Session = Depends(get_db)):
    updated = CrudControllers.update_Libro(db, libro_id, libro)
    if not updated:
        raise HTTPException(status_code=404, detail="Libro no encontrado")
    return updated

@router.delete("/libros/{libro_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_libro(libro_id: int, db: Session = Depends(get_db)):
    deleted = CrudControllers.delete_Libro(db, libro_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Libro no encontrado")

Archivo principal main.py (`main.py`)

from fastapi import FastAPI
from routes.routes import router

app = FastAPI()

#IMPORTANT: Aquí referenciamos el archivo "database" no la variable "db"
from database import database

def run():
    pass
if __name__ == '__main__':
    database.Base.metadata.create_all(database.engine)
    run()

@app.get("/")
async def root():
    return {"message": "Hello World"}

app.include_router(router, prefix="/v1")

8. Probando las Rutas con la Documentación de FastAPI

FastAPI genera automáticamente documentación interactiva de tu API utilizando Swagger UI. Esto nos permite probar las rutas sin necesidad de usar herramientas externas como Postman.

Acceder a la documentación

Asegúrate de que tu servidor esté corriendo:

uvicorn main:app --reload

Abre tu navegador y ve a:
- Swagger UI: http://127.0.0.1:8000/docs (Permite editar las peticiones)
- ReDoc: http://127.0.0.1:8000/redoc (No permite editar las peticiones)

En Swagger UI podrás ver todas las rutas disponibles, sus métodos, parámetros y modelos de request/response.

Lista de Rutas y Ejemplos

Ruta	Método	Descripción	Request Body (JSON)	Respuesta Ejemplo
`/v1/libros/`	POST	Crear un nuevo libro	`{ "title": "El Quijote", "description": "Novela de Cervantes" }`	`{ "id": 1, "title": "El Quijote", "description": "Novela de Cervantes" }`
`/v1/libros/`	GET	Obtener todos los libros	N/A	`[{"id": 1, "title": "El Quijote", "description": "Novela de Cervantes"}]`
`/v1/libros/{libro_id}`	GET	Obtener un libro por ID	N/A	`{ "id": 1, "title": "El Quijote", "description": "Novela de Cervantes" }`
`/v1/libros/{libro_id}`	PUT	Actualizar un libro por ID	`{ "title": "Don Quijote", "description": "Clásico de la literatura" }`	`{ "id": 1, "title": "Don Quijote", "description": "Clásico de la literatura" }`
`/v1/libros/{libro_id}`	DELETE	Eliminar un libro por ID	N/A	`204 No Content`

Cómo usar Swagger UI

Haz click en la ruta que quieras probar.
Si la ruta requiere parámetros (path params) o un request body, completa los campos que se muestran.
Haz click en "Execute" para enviar la petición.
Revisa la respuesta que aparece en la sección Response Body.

Nota: Las rutas están prefijadas con /v1 porque en main.py incluimos el router con app.include_router(router, prefix="/v1").

Ejemplo de prueba manual

Crear un libro

Ruta: POST /v1/libros/
Body:

{
  "title": "Cien años de soledad",
  "description": "Novela de Gabriel García Márquez"
}

Resultado esperado:

{
  "id": 2,
  "title": "Cien años de soledad",
  "description": "Novela de Gabriel García Márquez"
}

Obtener todos los libros

Ruta: GET /v1/libros/
No requiere body
Resultado esperado:

[
  { "id": 1, "title": "El Quijote", "description": "Novela de Cervantes" },
  { "id": 2, "title": "Cien años de soledad", "description": "Novela de Gabriel García Márquez" }
]

Actualizar un libro

Ruta: PUT /v1/libros/2
Body:

{
  "title": "Cien años de soledad - Edición revisada",
  "description": "Novela clásica de Gabriel García Márquez"
}

Resultado esperado:

{
  "id": 2,
  "title": "Cien años de soledad - Edición revisada",
  "description": "Novela clásica de Gabriel García Márquez"
}

Eliminar un libro

Ruta: DELETE /v1/libros/2
No requiere body
Resultado esperado: 204 No Content

9. Despliegue en Producción

Elegir un proveedor de hosting (por ejemplo, Heroku, DigitalOcean, AWS)
Configurar variables de entorno para la base de datos y otras configuraciones sensibles
Usar Gunicorn como servidor WSGI para producción
Configurar un servidor proxy inverso como Nginx (opcional, pero recomendado)
Implementar HTTPS para seguridad

Name		Name	Last commit message	Last commit date
Latest commit History 10 Commits
README.md		README.md

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Repository files navigation

Cómo Crear una Aplicación CRUD con FastAPI

Índice

1. Introducción

2. Estructura de Archivos y Carpetas

3. Instalación y Configuración

Instalar FastAPI y dependencias

Creamos nuestra conexión en Workbech

Comprobamos que estamos correctamente conectados a la base de datos

4. Conceptos Fundamentales

5. Flujo de Trabajo en FastAPI

6. Comandos Esenciales

7. Ejemplo Completo: Aplicación CRUD

Modelos (`models/libro_model.py`)

Schemas (`schemas/libro_schema.py`)

Operaciones CRUD (`controllers/libro_controller.py`)

Rutas de la API (`routes/routes.py`)

Archivo principal main.py (`main.py`)

8. Probando las Rutas con la Documentación de FastAPI

Acceder a la documentación

Lista de Rutas y Ejemplos

Cómo usar Swagger UI

Ejemplo de prueba manual

Crear un libro

Obtener todos los libros

Actualizar un libro

Eliminar un libro

9. Despliegue en Producción

10. Recursos Adicionales

About

Uh oh!

Releases

Packages

Factoria-F5-madrid/CRUD-FASTAPI

Folders and files

Latest commit

History

Repository files navigation

Cómo Crear una Aplicación CRUD con FastAPI

Índice

1. Introducción

2. Estructura de Archivos y Carpetas

3. Instalación y Configuración

Instalar FastAPI y dependencias

Creamos nuestra conexión en Workbech

Comprobamos que estamos correctamente conectados a la base de datos

4. Conceptos Fundamentales

5. Flujo de Trabajo en FastAPI

6. Comandos Esenciales

7. Ejemplo Completo: Aplicación CRUD

Modelos (models/libro_model.py)

Schemas (schemas/libro_schema.py)

Operaciones CRUD (controllers/libro_controller.py)

Rutas de la API (routes/routes.py)

Archivo principal main.py (main.py)

8. Probando las Rutas con la Documentación de FastAPI

Acceder a la documentación

Lista de Rutas y Ejemplos

Cómo usar Swagger UI

Ejemplo de prueba manual

Crear un libro

Obtener todos los libros

Actualizar un libro

Eliminar un libro

9. Despliegue en Producción

10. Recursos Adicionales

About

Topics

Resources

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Modelos (`models/libro_model.py`)

Schemas (`schemas/libro_schema.py`)

Operaciones CRUD (`controllers/libro_controller.py`)

Rutas de la API (`routes/routes.py`)

Archivo principal main.py (`main.py`)

Packages