Tutorial: Integración de FastAPI

Arquitectura sin bloqueo

La arquitectura asincrónica y sin bloqueos de FastAPI le permite gestionar miles de solicitudes simultáneas sin depender de subprocesos múltiples. Utiliza Python. asyncio Bucle de eventos para gestionar la concurrencia mediante corrutinas. 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.

Funciones e integraciones integradas

FastAPI tiene varias integraciones y funciones integradas, incluidos los siguientes componentes:

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

• OpenAPI y esquema JSON para documentación de API interactiva generada automáticamente

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

• Compatibilidad con autenticación y autorización OAuth2 y 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úrese de tener los siguientes componentes instalados y configurados antes de comenzar este tutorial:

• Python v3.9 - 3.11

• Un clúster de MongoDB Atlas. Para aprender a configurar un clúster, consulte 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"

Conecte la aplicación a su clúster

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

Utilice este código para realizar las siguientes acciones:

1. Conéctese a su clúster MongoDB Atlas utilizando el AsyncMongoClient() método con la MONGODB_URL variable de entorno y especificando la base de datos college denominada.

2. Crea 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 AsyncMongoClient() llamada incluye un server_api parámetro para habilitar la función API estable en su conexión. Asegúrese de que su implementación utilice servidores MongoDB con la versión 5.0 o posterior. Para obtener más información, consulte la sección API estable de esta guía.

Revisar los modelos de base de datos

Esta aplicación tiene tres modelos: StudentModel, UpdateStudentModel y StudentCollection. Estos modelos están definidos 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 comprende el funcionamiento de la aplicación, puede empezar a usar sus endpoints. Siga estos pasos para enviar solicitudes a sus endpoints y ver los resultados:

1

Inicie 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.

Haga clic en el punto final al que desea enviar una solicitud, complete los parámetros necesarios y haga clic en ExecuteEn el punto final Show Student, puede ver los datos de muestra definidos en la variable StudentModel.model_config.

Tip

Utilice curl para probar su API

Si prefiere utilizar la línea de comandos, puede enviar una solicitud mediante 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 POST, PUT y DELETE exitosas accediendo a su base de datos college en Atlas. También puede ver los resultados utilizando el punto de conexión List Students.

Próximos pasos

Ahora que comprende cómo FastAPI se integra con MongoDB y el controlador asíncrono de PyMongo, puede agregar más funciones. La siguiente lista incluye algunas sugerencias para 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 Medio Ambiente

Almacenar credenciales confidenciales directamente en el código de la aplicación podría causar una exposición accidental en los sistemas de control de versiones. En su lugar, utilice variables de entorno para almacenar cadenas de conexión, claves de 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 .envrc archivo 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 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 de producción, utilice un Sistema de Gestión de Claves (KMS). También puede usar soluciones KMS en su entorno de desarrollo para mejorar la seguridad.

Ciclo de vida de la conexión a la base de datos

Gestione correctamente el ciclo de vida de sus conexiones de MongoDB inicializando y cerrando las conexiones de cliente en respuesta a los eventos de inicio y cierre de la aplicación FastAPI. Este enfoque es más robusto que inicializar las conexiones a nivel de módulo.

Conecte su cliente MongoDB al objeto de aplicación FastAPI para que sea accesible a todas las funciones de operación de rutas en su código. Esto elimina la necesidad de variables globales y simplifica la gestión de dependencias.

Utilice los eventos de vida útil de FastAPI para gestionar la administración de 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 puntos finales de FastAPI que proporcionan datos JSON a MongoDB pueden contener tipos ObjectId y Binary, que no son compatibles de forma nativa con JSON.

Puedes serializar ObjectId de varias maneras:

• 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 de Pydantic: utilice las anotaciones de tipo de Pydantic para manejar la conversión automática, como se muestra en el PyObjectId tipo 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 le permite modificar las estructuras de los documentos de MongoDB sin cambiar la interfaz de su 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}'"
        )