Tutorial: Integración con FastAPI

Arquitectura no bloqueante

La arquitectura asincrónica y no bloqueante de FastAPI le permite gestionar miles de solicitudes simultáneas sin depender de la multiproceso. Utiliza la de Python asyncio bucle de eventos para gestionar la concurrencia a través de co-rutinas. Esto permite que la aplicación suspenda una solicitud mientras espera el retorno de las consultas. Esto reduce el uso de memoria y puede reducir la latencia.

Funcionalidades e integraciones integradas

FastAPI cuenta con varias integraciones y funcionalidades incorporadas, incluyendo los siguientes componentes:

• Pydantic para el análisis automático de datos, validación y serialización

• OpenAPI y JSON Schema para documentación de API autogenerada e interactiva

• Un sistema ligero de inyección de dependencias Pythonic para un código limpio y comprobable

• Soporte para OAuth2 y autenticación y autorización JWT

Estas funcionalidades minimizan el código repetitivo para simplificar el desarrollo. Para obtener más información sobre las capacidades de FastAPI, consulta el Sitio web de FastAPI.

Requisitos previos

Asegúrate de tener instalados y configurados los siguientes componentes antes de comenzar este tutorial:

• Python v3.9 - 3.11

• Un clúster de MongoDB Atlas. Para aprender a configurar un clúster, consulte la Guía de introducción para obtener más información.

Configuración

Instala las dependencias, conéctate a MongoDB y arranca tu servidor FastAPI:

git clone git@github.com:mongodb-developer/mongodb-pymongo-fastapi.git
cd mongodb-pymongo-fastapi

cd mongodb-pymongo-fastapi
pip install -r requirements.txt

export MONGODB_URL="mongodb+srv://<username>:<password>@<url>/<db>?retryWrites=true&w=majority"

Conecta la aplicación a tu clúster

Todo el código de la aplicación de ejemplo se almacena en el archivo app.py en el mongodb-pymongo-fastapi repositorio de GitHub.

Usa este código para realizar las siguientes acciones:

1. Conecte a su clúster de MongoDB Atlas usando el método AsyncMongoClient() con la variable de entorno MONGODB_URL y especificando la base de datos llamada college.

2. Cree un puntero a la base de datos college.

3. Cree un puntero a la colección students

client = AsyncMongoClient(os.environ["MONGODB_URL"],server_api=pymongo.server_api.ServerApi(version="1", strict=True,deprecation_errors=True))
db = client.get_database("college")
student_collection = db.get_collection("students")

La llamada AsyncMongoClient() incluye un parámetro server_api para habilitar la funcionalidad Stable API en tu conexión. Asegúrate de que tu implementación use servidores MongoDB con la versión 5.0 o posterior. Para más información, consulta la sección Stable API de esta guía.

Revisar los Modelos de base de datos

Esta aplicación tiene tres modelos, el StudentModel, el UpdateStudentModel y el StudentCollection. Estos modelos se definen en el archivo app.py.

Todos los modelos de la aplicación se basan en el Pydantic BaseModel, que proporciona validación básica de tipos, análisis y serialización JSON y manejo básico de errores.

# Represents an ObjectId field in the database.
# It will be represented as a `str` on the model so that it can be serialized to JSON.
PyObjectId = Annotated[str, BeforeValidator(str)]
class StudentModel(BaseModel):
    """
    Container for a single student record.
    """
    # The primary key for the StudentModel, stored as a `str` on the instance.
    # This will be aliased to ``_id`` when sent to MongoDB,
    # but provided as ``id`` in the API requests and responses.
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    name: str = Field(...)
    email: EmailStr = Field(...)
    course: str = Field(...)
    gpa: float = Field(..., le=4.0)
    model_config = ConfigDict(
        populate_by_name=True,
        arbitrary_types_allowed=True,
        json_schema_extra={
            "example": {
                "name": "Jane Doe",
                "email": "jdoe@example.com",
                "course": "Experiments, Science, and Fashion in Nanophotonics",
                "gpa": 3.0,
            }
        },
    )

class UpdateStudentModel(BaseModel):
    """
    A set of optional updates to be made to a document in the database.
    """
    name: Optional[str] = None
    email: Optional[EmailStr] = None
    course: Optional[str] = None
    gpa: Optional[float] = None
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_encoders={ObjectId: str},
        json_schema_extra={
            "example": {
                "name": "Jane Doe",
                "email": "jdoe@example.com",
                "course": "Experiments, Science, and Fashion in anophotonics",
                "gpa": 3.0,
            }
        },
    )

class StudentCollection(BaseModel):
"""
A container holding a list of `StudentModel` instances
"""
students: List[StudentModel]

Revisar las rutas de la aplicación

La siguiente tabla describe las rutas de aplicación definidas en esta aplicación:

Estas rutas se definen en el archivo app.py como se describe en las siguientes secciones:

@app.post(
    "/students/",
    response_description="Add new student",
    response_model=StudentModel,
    status_code=status.HTTP_201_CREATED,
    response_model_by_alias=False,
)
async def create_student(student: StudentModel = Body(...)):
    """
    Insert a new student record.
    A unique ``id`` will be created and provided in the response.
    """
    new_student = student.model_dump(by_alias=True, exclude=["id"])
    result = await student_collection.insert_one(new_student)
    new_student["_id"] = result.inserted_id
    return new_student

@app.get(
    "/students/",
    response_description="List all students",
    response_model=StudentCollection,
    response_model_by_alias=False,
)
async def list_students():
    """
    List all the student data in the database.
    The response is unpaginated and limited to 1000 results.
    """
return StudentCollection(students=await student_collection.find().to_list(1000))

@app.get(
    "/students/{id}",
    response_description="Get a single student",
    response_model=StudentModel,
    response_model_by_alias=False,
)
async def show_student(id: str):
    """
    Get the record for a specific student, looked up by ``id``.
    """
    if (
        student := await student_collection.find_one({"_id": ObjectId(id)})
    ) is not None:
        return student
    raise HTTPException(status_code=404, detail="Student {id} not found")

@app.put(
    "/students/{id}",
    response_description="Update a student",
    response_model=StudentModel,
    response_model_by_alias=False,
)
async def update_student(id: str, student: UpdateStudentModel = Body(...)):
    """
    Update individual fields of an existing student record.
    Only the provided fields will be updated.
    Any missing or `null` fields will be ignored.
    """
    student = {
        k: v for k, v in student.model_dump(by_alias=True).items() if v is not None
    }
    if len(student) >= 1:
        update_result = await student_collection.find_one_and_update(
            {"_id": ObjectId(id)},
            {"$set": student},
            return_document=ReturnDocument.AFTER,
        )
        if update_result is not None:
            return update_result
        else:
            raise HTTPException(status_code=404, detail=f"Student {id} not found")
    # The update is empty, so return the matching document:
    if (existing_student := await student_collection.find_one({"_id": ObjectId(id)})) is not None:
        return existing_student
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

@app.delete("/students/{id}", response_description="Delete a student")
async def delete_student(id: str):
    """
    Remove a single student record from the database.
    """
    delete_result = await student_collection.delete_one({"_id": ObjectId(id)})
    if delete_result.deleted_count == 1:
        return Response(status_code=status.HTTP_204_NO_CONTENT)
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

Probar la API

Ahora que entiendes cómo funciona la aplicación, puedes empezar a usar tus puntos finales. Sigue los siguientes pasos para enviar solicitudes a tus endpoints y ver los resultados:

1

Arrancar su servidor FastAPI

Ejecute el siguiente código en su terminal para iniciar su servidor FastAPI:

uvicorn app:app --reload

2

Enviar solicitudes

Vaya a http://:127.0.0.1 8000/docs. La siguiente imagen muestra la interfaz generada por el servidor FastAPI.

Haz clic en el endpoint al que deseas enviar una solicitud, rellena los parámetros necesarios y haz clic ExecuteEn el punto final Show Student, puede ver los datos de muestra definidos en la variable StudentModel.model_config.

Tip

Usa curl para probar tu API

Si prefieres utilizar la línea de comandos, puedes enviar una solicitud usando el comando curl, como este:

curl -X POST "http://127.0.0.1:8000/students/" \
    -H "Content-Type: application/json" \
    -d '{"name": "Jane Doe", "email": "jdoe@example.com", "course": "Physics", "gpa": 3.8}'

3

Vea los resultados de sus solicitudes

Puede ver los resultados de las solicitudes exitosas de POST, PUT y DELETE navegando a su base de datos college en Atlas. También puedes ver tus resultados usando el endpoint List Students.

Próximos pasos

Ahora que tienes una comprensión básica de cómo FastAPI se integra con MongoDB y el driver asíncrono de PyMongo, puedes añadir más funcionalidades. La siguiente lista incluye algunas sugerencias sobre cómo podrías aportar a esta aplicación:

• Validación de datos adicionales

• Inyección de dependencia

• Pruebas con TestClient

• Manejo de errores y respuestas personalizadas

• Autenticación y autorización

Seguridad y entorno

Almacenar credenciales sensibles directamente en el código de la aplicación podría provocar una exposición accidental en los sistemas de control de versiones. En su lugar, use variables de entorno para almacenar cadenas de conexión, claves API y otra información confidencial.

Para entornos de desarrollo, puede utilizar las siguientes herramientas para administrar las variables de entorno:

• direnv: Carga automáticamente variables de entorno desde un archivo .envrc al ingresar a un directorio

• python-dotenv: carga variables de entorno desde un .env archivo dentro de su aplicación Python

• envdir: Establece variables de entorno a partir de archivos en un directorio

• Honcho: carga variables de entorno desde archivos de configuración

Independientemente de la herramienta de control de variables de entorno que elija, agregue cualquier archivo que contenga secretos (como .env, .envrc o similares) a su archivo .gitignore para evitar que se confirmen en el control de versiones.

Para implementaciones en producción, utiliza un sistema de gestión de claves (KMS). También puede utilizar soluciones KMS en su entorno de desarrollo para mejorar la seguridad.

Ciclo de Vida de la Conexión de la Base de Datos

Gestiona adecuadamente el ciclo de vida de tu conexión a MongoDB iniciando y cerrando las conexiones del cliente en respuesta a los eventos de inicio y apagado de la aplicación FastAPI. Este enfoque es más robusto que iniciar conexiones a nivel de módulo.

Conecta tu cliente MongoDB al objeto de aplicación de FastAPI para hacerlo accesible a todas las funciones de operaciones de ruta en toda tu base de código. Esto elimina la necesidad de variables globales y simplifica la gestión de dependencias.

Usa los eventos de ciclo de vida de FastAPI para gestionar la conexión:

import os  
from contextlib import asynccontextmanager 
from logging import info
from fastapi import FastAPI
from pymongo import AsyncMongoClient
@asynccontextmanager
async def db_lifespan(app: FastAPI):
    # Startup
    app.mongodb_client = AsyncMongoClient(os.environ["MONGODB_URL"])
    app.database = app.mongodb_client.get_default_database()
    ping_response = await app.database.command("ping")
    if int(ping_response["ok"]) != 1:
        raise Exception("Problem connecting to database cluster.")
    else:
        info("Connected to database cluster.")
    yield
    # Shutdown
    await app.mongodb_client.close()
app: FastAPI = FastAPI(lifespan=db_lifespan)

Manejo de tipos BSON vs JSON

Los endpoints de FastAPI que proporcionan datos JSON a MongoDB podrían contener tipos ObjectId y Binary, que no cuentan con soporte nativo en JSON.

Puedes serializar ObjectId de varias formas:

• Usar IDs basados en cadenas: Reemplaza ObjectId con valores de cadena compatibles con JSON para campos _id. Este enfoque puede funcionar bien para nuevas aplicaciones, pero puede no ser adecuado si tienes datos existentes.

• Conversión automática: convierte ObjectId valores en representaciones de cadenas al serializar en JSON y analiza cadenas nuevamente como ObjectId al procesar datos entrantes.

• Integración Pydantic: Utiliza las anotaciones de tipo de Pydantic para gestionar la conversión automática, como se demuestra en el tipo PyObjectId de este tutorial.

• Mapeadores de documentos de objetos (ODM): utilice un ODM como Beanie, que asume que los _id campos usan ObjectId de manera predeterminada y maneja automáticamente la serialización y la deserialización.

Modelado de datos

Separe el esquema de su base de datos del esquema de su API para mantener la flexibilidad. Esto permite modificar las estructuras de los documentos de MongoDB sin cambiar su interfaz de API.

Proporcione una clase Pydantic como response_model para su operación de ruta para convertir, validar, documentar y filtrar los campos de los datos BSON devueltos. El siguiente ejemplo de código muestra cómo implementar esta técnica:

# A Pydantic class modelling the *response* schema.
class Profile(BaseModel):
    """
    A profile for a single user.
    """
    id: Optional[str] = Field(
        default=None, description="MongoDB document ObjectID", alias="_id"
    )
    username: str
    residence: str
    current_location: List[float]
# A path operation that returns a Profile object as JSON.
app = FastAPI()
@app.get(
    "/profiles/{profile_id}",
    response_model=Profile,  # Tells FastAPI that the returned object must match the Profile schema.
)
async def get_profile(profile_id: str) -> Mapping[str, Any]:
    # Uses response_model to automatically convert, validate, and document
    # the returned dict without manual Profile object creation.
    profile = await app.profiles.find_one({"_id": profile_id})
    if profile is not None:
        # Return BSON document. FastAPI converts it automatically.
        return profile
    else:
        raise HTTPException(
            status_code=404, detail=f"No profile with id '{profile_id}'"
        )