Tutorial: integração com a FastAPI

Arquitetura não bloqueante

A arquitetura assíncrona e não bloqueante da FastAPI permite que ela lide com milhares de solicitações simultâneas sem depender de várias threads. Ele usa o loop de eventos asyncio do Python para gerenciar a simultaneidade por meio de corrotinas. Isso permite que o aplicação suspenda uma solicitação enquanto aguarda o retorno das queries. Isso reduz o uso de memória e pode reduzir a latência.

Recursos e integrações integrados

A FastAPI tem várias integrações e recursos integrados, incluindo os seguintes componentes:

• Pydentic para análise, validação e serialização automáticas de dados

• OpenAPI e JSON schema para documentação de API interativa e gerada automaticamente

• Um sistema leve de injeção de dependência Pythonic para código limpo e testável

• Suporte para autenticação e autorização de token web OAuth2 e JSON

Esses recursos minimizam o código boilerplate para simplificar o desenvolvimento. Para obter mais informações sobre os recursos da FastAPI, consulte o site da FastAPI.

Pré-requisitos

Certifique-se de ter os seguintes componentes instalados e configurados antes de iniciar este tutorial:

• Python v3.9 - 3.11

• Um cluster MongoDB Atlas . Para saber como configurar um cluster, consulte o guiaComeçando do para mais informações.

Configuração

Instale dependências, conecte-se ao MongoDB e inicie seu 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 o aplicativo ao seu cluster

Todo o código do aplicação de exemplo é armazenado no arquivo app.py no repositório mongodb-pymongo-fastapi do Github.

Use este código para executar as seguintes ações:

1. Conecte-se ao MongoDB Atlas cluster usando o método AsyncMongoClient() com a variável de ambiente MONGODB_URL e especificando o banco de dados chamado college.

2. Crie um ponteiro para o banco de dados college .

3. Criar um ponteiro para a coleção 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")

A chamada AsyncMongoClient() inclui um parâmetro server_api para habilitar o recurso de API estável em sua conexão. Certifique-se de que seu sistema use servidores MongoDB com versão 5.0 ou posterior. Para obter mais informações, consulte a seção API estável deste guia.

Revise os modelos de banco de dados

Este aplicação tem três modelos, o StudentModel, o UpdateStudentModel e o StudentCollection. Estes modelos são definidos no arquivo app.py.

Todos os modelos no aplicação se baseiam no Pydantic BaseModel, que fornece validação básica de tipo, análise e serialização JSON e tratamento básico de erros.

# 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]

Revise as rotas de aplicativos

A tabela a seguir descreve as rotas do aplicação definidas neste aplicação:

Estas rotas são definidas no arquivo app.py, conforme descrito nas seguintes seções:

@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")

Teste a API

Agora que você entende como o aplicação funciona, pode começar a usar seus endpoints. Use as etapas a seguir para enviar solicitações para seus endpoints e ver os resultados:

1

Inicie seu servidor FastAPI

Execute o seguinte código no seu terminal para iniciar o servidor FastAPI :

uvicorn app:app --reload

2

Enviar solicitações

Navegue até http://127.0.0.1:8000/Docs. A imagem seguinte mostra a interface gerada pelo servidor FastAPI.

Clique no endpoint para o qual deseja enviar uma solicitação, preencha todos os parâmetros necessários e clique em Execute. No endpoint Show Student , você pode ver os dados de amostra definidos na variável StudentModel.model_config .

Dica

Use o curl para testar sua API

Se preferir usar a linha de comando, você pode enviar uma solicitação usando o 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

Veja os resultados das suas solicitações

Você pode ver os resultados de solicitações bem-sucedidas de POST, PUT e DELETE navegando até seu banco de dados college no Atlas. Você também pode ver seus resultados usando o ponto de extremidade List Students.

Próximos passos

Agora que você tem uma compreensão básica de como a FastAPI se integra ao MongoDB e ao driver PyMongo Async, você pode adicionar mais recursos. A lista a seguir inclui algumas sugestões de como você pode adicionar a este aplicação:

• Validação de dados adicionais

• Injeção de dependência

• Testando com TestClient

• Tratamento de erros e respostas personalizadas

• Autenticação e autorização

Segurança e ambiente

Armazenar credenciais confidenciais diretamente no código do aplicação pode causar exposição acidental nos sistemas de controle de versão. Em vez disso, use variáveis de ambiente para armazenar connection strings, chaves de API e outras informações confidenciais.

Para ambientes de desenvolvimento, você pode usar as seguintes FERRAMENTAS para gerenciar variáveis de ambiente:

• direnv: Carrega automaticamente variáveis de ambiente de um arquivo .envrc ao inserir um diretório

• python-dotenv: Carrega variáveis de ambiente de um arquivo .env dentro do seu aplicação Python

• envdir: define variáveis de ambiente a partir de arquivos em um diretório

• Honcho: Carrega variáveis de ambiente de arquivos de configuração

Independentemente da ferramenta de controle de variável de ambiente que você escolher, adicione quaisquer arquivos que contenham segredos (como .env, .envrc ou semelhantes) ao arquivo .gitignore para evitar que eles sejam comprometidos com o controle de versão.

Para sistemas de produção, use um Sistema de Gerenciamento de Chaves (KMS). Você também pode usar soluções KMS em seu ambiente de desenvolvimento para aumentar a segurança.

Ciclo de vida da conexão do banco de dados

Gerencie adequadamente o ciclo de vida da conexão do MongoDB inicializando e fechando as conexões do cliente em resposta aos eventos de inicialização e desligamento do aplicação FastAPI. Essa abordagem é mais robusta do que inicializar conexões no nível do módulo.

Anexe seu cliente MongoDB ao objeto de aplicação FastAPI para torná-lo acessível a todas as funções de operação de caminho em toda a sua base de código. Isso elimina a necessidade de variáveis globais e simplifica o gerenciamento de dependência.

Use os eventos de duração do FastAPI para lidar com o gerenciamento de conexões:

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)

Manipulação de tipos de BSON vs JSON

Os endpoints FastAPI que fornecem dados JSON ao MongoDB podem conter tipos ObjectId e Binary, que não são nativamente suportados em JSON.

Você pode serializar o ObjectId de várias maneiras:

• Usar IDs baseados em string: substitua ObjectId por valores de string compatíveis com JSON para campos _id. Essa abordagem pode funcionar bem para novos aplicativos, mas pode não ser adequada se você tiver dados existentes.

• Conversão automática: converta valores ObjectId em representações de string ao serializar para JSON e analise strings de volta para ObjectId ao processar dados de entrada.

• Integração com o Pydentic: use as anotações de tipo do Pydantial para lidar com a conversão automática, conforme demonstrado no tipo PyObjectId deste tutorial.

• Mapeadores de documentos de objetos (ODMs): use um ODM como Beanie, que pressupõe que os campos _id usem ObjectId por padrão e lida automaticamente com a serialização e a desserialização.

Modelagem de dados

Separe seu esquema de banco de dados do esquema de API para manter a flexibilidade. Isso permite que você modifique as estruturas de documento do MongoDB sem alterar a interface da API.

Forneça uma classe Pydantial como response_model para que sua operação de caminho converta, valide, documento e filtre os campos dos dados BSON retornados. O seguinte exemplo de código mostra como 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}'"
        )