En esta guía, te mostraré cómo crear una API REST completa utilizando el framework Python FastAPI. Nuestro proyecto consistirá en un sistema de gestión de tareas (TODO API) con funcionalidades CRUD (Create, Read, Update, Delete), autenticación JWT, validación de datos con Pydantic y almacenamiento de datos en una base de datos SQLite.
1. Setup del entorno e instalación
Para comenzar, necesitamos instalar los siguientes paquetes:
- FastAPI
- Pydantic
- sqlite3
- uvicorn (para ejecutar nuestra API)
Puedes instalarlos mediante pip:
pip install fastapi pydantic sqlite3 uvicornCrea un nuevo directorio para tu proyecto y agrega un archivo main.py vacío. Este archivo será el punto de entrada de nuestra aplicación.
2. Primera ruta y «Hello World»
En este paso, creamos nuestra primera ruta en la API. Agrega el siguiente código a main.py:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}Esto define una ruta GET en / que devuelve un objeto JSON con un mensaje de bienvenida. Ejecuta la aplicación con:
uvicorn main:app --host 0.0.0.0 --port 8000Luego, accede a http://localhost:8000/ para ver el resultado.
3. Modelos de datos con Pydantic
Para definir nuestros modelos de datos, crearemos un archivo models.py y agregaremos los siguientes modelos:
from pydantic import BaseModel
class Task(BaseModel):
id: int
title: str
description: str
class User(BaseModel):
id: int
username: str
password: strEstos modelos definen las estructuras de los datos que se van a utilizar en nuestra API.
4. Conexión a base de datos
Creamos un archivo database.py y agregamos la siguiente conexión a una base de datos SQLite:
import sqlite3
class Database:
def __init__(self):
self.conn = sqlite3.connect("tasks.db")
self.cursor = self.conn.cursor()
def create_table(self):
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
description TEXT NOT NULL
);
""")
def insert_task(self, task: Task):
self.cursor.execute(
"INSERT INTO tasks (title, description) VALUES (:title, :description)",
{"title": task.title, "description": task.description}
)
self.conn.commit()
def get_tasks(self):
self.cursor.execute("SELECT * FROM tasks")
return [Task(**row) for row in self.cursor.fetchall()]Este archivo define una clase Database que se encarga de conectarse a la base de datos, crear una tabla para almacenar tareas y insertar y obtener tareas.
5. Endpoints CRUD con validaciones
Agregamos los siguientes endpoints a main.py:
from fastapi import FastAPI, HTTPException
from pydantic import ValidationError
from models import Task
from database import Database
app = FastAPI()
db = Database()
db.create_table()
@app.post("/tasks")
async def create_task(task: Task):
try:
db.insert_task(task)
return {"message": "Task created successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/tasks")
async def get_tasks():
tasks = db.get_tasks()
return [{"id": task.id, "title": task.title, "description": task.description} for task in tasks]
@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
try:
task = next((task for task in db.get_tasks() if task.id == task_id), None)
return {"id": task.id, "title": task.title, "description": task.description}
except StopIteration:
raise HTTPException(status_code=404, detail="Task not found")
@app.put("/tasks/{task_id}")
async def update_task(task_id: int, task: Task):
try:
db.insert_task(task)
return {"message": "Task updated successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
try:
db.cursor.execute("DELETE FROM tasks WHERE id = :id", {"id": task_id})
db.conn.commit()
return {"message": "Task deleted successfully"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Estos endpoints permiten crear, leer, actualizar y eliminar tareas. Se utilizan las clases de modelo Task y la clase Database para interactuar con la base de datos.
6. Sistema de autenticación
Creamos un archivo auth.py y agregamos el siguiente código:
from fastapi import FastAPI, HTTPException
from pydantic import ValidationError
from models import User
app = FastAPI()
@app.post("/login")
async def login(username: str, password: str):
try:
user = next((user for user in users if user.username == username and user.password == password), None)
if user:
return {"access_token": "secret"}
else:
raise HTTPException(status_code=401, detail="Invalid credentials")
except StopIteration:
raise HTTPException(status_code=401, detail="Invalid credentials")
@app.get("/protected")
async def protected_route(access_token: str):
if access_token == "secret":
return {"message": "Hello, authenticated user!"}
else:
raise HTTPException(status_code=403, detail="Forbidden")Este sistema de autenticación utiliza endpoints para loguearse y obtener un token de acceso. El endpoint /protected solo permite acceder a aquellos que tienen el token correcto.
7. Manejo de errores HTTP
En lugar de manejar errores con try/except, podemos utilizar las excepciones HTTP definidas en FastAPI. Por ejemplo, en los endpoints CRUD, podemos lanzar una excepción HTTP 404 si no se encuentra la tarea:
@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
try:
task = next((task for task in db.get_tasks() if task.id == task_id), None)
return {"id": task.id, "title": task.title, "description": task.description}
except StopIteration:
raise HTTPException(status_code=404, detail="Task not found")8. Testing de la API
Creamos un archivo test_api.py y agregamos los siguientes tests con pytest:
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
return TestClient(app)
def test_root(client):
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_task(client, task: Task):
response = client.post("/tasks", json=task.dict())
assert response.status_code == 201
assert response.json() == {"message": "Task created successfully"}
def test_get_tasks(client):
response = client.get("/tasks")
assert response.status_code == 200
assert response.json() == []
def test_login(client, username: str, password: str):
response = client.post("/login", json={"username": username, "password": password})
assert response.status_code == 200
assert response.json() == {"access_token": "secret"}
def test_protected_route(client):
response = client.get("/protected")
assert response.status_code == 403Estos tests verifican que los endpoints estén funcionando correctamente.
9. Documentación y deployment
Para generar documentación automática, podemos utilizar la biblioteca Swagger UI. Agregamos el siguiente código a main.py:
from fastapi import FastAPI, OpenAPI
from uvicorn.main import run
app = FastAPI()
app.openapi = OpenAPI(
title="Todo API",
description="A simple Todo API built with FastAPI and SQLite",
version="1.0"
)
if __name__ == "__main__":
run(app, host="0.0.0.0", port=8000)Para deployment, podemos crear un archivo docker-compose.yml y agregar el siguiente contenido:
version: "3"
services:
app:
build: .
ports:
- "8000:8000"
environment:
DATABASE_URL: sqlite:///tasks.db
depends_on:
- db
db:
image: postgres
volumes:
- tasks.db:/var/lib/postgresql/data/tasks.db
volumes:
tasks.db:Este archivo define un servicio app que se encarga de construir la imagen del contenedor y expone el puerto 8000. El servicio db se encarga de crear una base de datos PostgreSQL y montarla en el volumen tasks.db.
10. README detallado
Agrega un archivo README.md con información detallada sobre cómo ejecutar la aplicación, como instalar los paquetes necesarios, cómo configurar la base de datos y cómo correr los tests.
Conclusión
¡Eso es todo! Con este proyecto, hemos creado una API REST completa utilizando FastAPI y Pydantic. Espero que hayas disfrutado de esta guía paso a paso y que te haya sido útil para aprender sobre APIs y desarrollo web con Python.
Deja una respuesta