Una API con Python y FastAPI

25 minPrincipiante

Con FastAPI y PyMongo basta para una API de pedidos que valida cada pedido y lo guarda en TeselaDB.

Una API de Python usa TeselaDB con PyMongo, el driver oficial de MongoDB. Aquí escribes con FastAPI la misma API de pedidos que en la guía de Express, que crea un pedido y lo lee.

Lo que vas a construir

Un servidor FastAPI en el puerto 8000 con POST /orders, que valida el pedido con Pydantic y lo guarda, y GET /orders/{order_id}, que lo devuelve o responde 404.

Requisitos

  • Python 3.9 o posterior.
  • Docker y mongosh, para arrancar TeselaDB en tu máquina y mirar los datos.

Prepara el entorno

Shell
docker run -d --name teseladb -p 27017:27017 viuzi/teseladb:latest \
  start --dbpath /data --insecure --insecure-allow-remote --bind_ip 0.0.0.0
export TESELADB_URI="mongodb://127.0.0.1:27017/shop"
mongosh "$TESELADB_URI" --quiet --eval 'db.runCommand({ ping: 1 }).ok'

El contenedor arranca un nodo de TeselaDB sin TLS ni usuarios, que basta para seguir la guía en tu máquina. Cuando mongosh responde 1, el nodo ya acepta conexiones. TESELADB_URI apunta a ese nodo y a la base shop; en un clúster de producción la cadena lleva TLS y usuario, como cuenta Conectar tu aplicación.

1. Crea el entorno

Shell
python3 -m venv .venv
. .venv/bin/activate
pip install fastapi uvicorn pymongo

2. Escribe la API

Guárdala como main.py. Las funciones de los endpoints son síncronas, así que FastAPI las ejecuta en su grupo de hilos y el cliente de PyMongo comparte sus conexiones entre ellas:

Python
import os
from datetime import datetime, timezone

from bson import ObjectId
from bson.errors import InvalidId
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from pymongo import MongoClient

client = MongoClient(os.environ["TESELADB_URI"])
orders = client.shop.orders
app = FastAPI()


class NewOrder(BaseModel):
    customer: str
    total: float = Field(ge=0)


@app.post("/orders", status_code=201)
def create_order(order: NewOrder):
    result = orders.insert_one({**order.model_dump(), "createdAt": datetime.now(timezone.utc)})
    return {"id": str(result.inserted_id)}


@app.get("/orders/{order_id}")
def read_order(order_id: str):
    try:
        order = orders.find_one({"_id": ObjectId(order_id)})
    except InvalidId:
        order = None
    if order is None:
        raise HTTPException(status_code=404)
    order["_id"] = str(order["_id"])
    return order

Probarlo

Arranca el servidor con uvicorn main:app. Después, en otra terminal, crea un pedido de Ana con un total de 30 y léelo con el id que devuelve la primera petición, que sustituye a <id> en la segunda orden:

Shell
curl -s -X POST localhost:8000/orders -H "content-type: application/json" -d '{"customer":"Ana","total":30}'
curl -s localhost:8000/orders/<id>

La primera orden devuelve {"id":"…"} y la segunda, el pedido con su fecha de creación. Un pedido con un total negativo responde 422 con el error de validación.

Limpiar

Para el servidor con Ctrl+C y borra los pedidos de prueba desde mongosh. Al final borra también el contenedor con la última orden: el nodo guarda los datos dentro, sin volumen, así que con él desaparece todo lo que escribiste durante la guía y el puerto 27017 queda libre:

JavaScript
db.getSiblingDB("shop").orders.deleteMany({ customer: "Ana" });
Shell
docker rm -f teseladb

Siguientes pasos