Como Instalar Fast API com MongoDB no Ubuntu 24.04

FastAPI é um framework web baseado em Python para criar serviços de API. É um framework moderno, rápido e de alto desempenho que suporta operações assíncronas.

Neste tutorial, você aprenderá como instalar o FastAPI com MongoDB no Ubuntu 24.04. Você também aprenderá como criar sua primeira API com operações CRUD usando o FastAPI e o banco de dados MongoDB.

Pré-requisitos

Para começar este guia, certifique-se de ter o seguinte:

• Um sistema Ubuntu 24.04
• Um usuário não-root com privilégios de administrador

Instalando o MongoDB

Antes de criar um novo projeto FastAPI, vamos instalar o servidor MongoDB em nosso sistema.

Primeiro, execute o comando abaixo para atualizar seu índice de pacotes e instalar ‘gnupg’ e ‘curl’ em seu sistema.

sudo apt update && sudo apt install gnupg curl -y

Execute o comando abaixo para adicionar a chave GPG para o servidor MongoDB.

curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | 
sudo gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg 
--dearmor

Adicione o repositório MongoDB usando o comando abaixo.

echo "deb [ arch=amd64,arm64 signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] https://repo.mongodb.org/apt/ubuntu noble/mongodb-org/8.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-8.0.list

Após adicionar o repositório, execute o seguinte comando para atualizar seu índice de pacotes e instalar o servidor MongoDB. Digite ‘Y’ para confirmar a instalação.

sudo apt update && sudo apt install mongodb-org

Quando a instalação estiver concluída, inicie e habilite o serviço MongoDB ‘mongod’ e, em seguida, verifique o status do serviço MongoDB para garantir que ele esteja em execução.

sudo systemctl enable --now mongod  
sudo systemctl status mongod

Você pode ver abaixo que o servidor MongoDB está em execução.

Além disso, você pode fazer login no servidor MongoDB com o comando ‘mongosh’ abaixo. Para sair, pressione Ctrl+d.

mongosh

Configurando Python e Ambiente Virtual

Com o MongoDB instalado, você instalará pacotes Python e configurará o diretório do projeto e o ambiente virtual.

Instale os módulos Python, Pip e Venv com o seguinte comando. Digite ‘Y’ para confirmar a instalação.

sudo apt install python3 python3-pip python3-venv

Uma vez que a instalação esteja concluída, faça login no seu usuário.

su - username

Agora crie um novo diretório ‘~/app‘ e entre nele. Este diretório será usado para armazenar seu projeto FastAPI.

mkdir -p ~/app; cd ~/app

Execute o comando abaixo para criar um novo ambiente virtual ‘venv‘ e ativá-lo. Com isso, seu prompt de shell se tornará como ‘(venv) user@hostname‘.

python3 -m venv .venv  
source .venv/bin/activate

A partir daqui, seu ambiente de trabalho deve estar no ambiente virtual ‘venv’. Você pode sair do ‘venv’ usando o comando abaixo.

deactivate

Criando Projeto FastAPI

Agora que você criou e ativou seu ambiente virtual Python, vamos instalar o FastAPI e criar a estrutura do projeto.

Com o comando ‘pip3’, execute-o para instalar os pacotes ‘fastapi’ e ‘uvicorn’.

pip3 install fastapi uvicorn

• O ‘fastapi’ é o principal framework web FastAPI para construir APIs em Python
• O ‘uvicorn’ é a implementação do servidor web ASGI (Asynchronous Server Gateway Interface) em Python.

Após a instalação ser concluída, crie novos arquivos e diretórios com o seguinte comando.

mkdir -p server/{models,routes}  
touch main.py server/{app.py,database.py} server/models/itemModels.py server/routes/item.py

Abaixo está nossa estrutura de projeto FastAPI.

server/app.py

Agora que seu projeto está pronto, vamos modificar o arquivo ‘server/app.py’, que é o aplicativo principal do seu projeto FastAPI.

Abra o arquivo ‘app.py‘ com seu editor de texto e copie o seguinte script.

from fastapi import FastAPI  
  
app = FastAPI()  
  
@app.get("/")  
async def root():  
 return {"message": "Hello FastAPI!"}

• Importa o módulo FastAPI para seu projeto e o vincula à variável ‘app’
• Cria uma nova função ‘root’ que retorna o ‘Hello FastAPI!’
• A função ‘root’ responde ao método GET na URL raiz
• ‘async’ marca sua função como uma função assíncrona e pode usar ‘await’ dentro de seu corpo quando chamada

main.py

Nesta seção, vamos modificar o arquivo ‘main.py’ que será usado para executar seu projeto FastAPI via ‘uvicorn’ (servidor web ASGI em Python).

Agora abra e modifique o script ‘main.py‘ e insira o seguinte código.

import uvicorn  
  
if __name__ == "__main__":  
 uvicorn.run("server.app:app", host="0.0.0.0", port=8080, reload=True)

• Importa o módulo ‘uvicorn’
• Quando o script ‘main.py’ é executado, ele carregará o módulo ‘app’ ou FastAPI dentro do ‘server/app.py’
• O FastAPI estará rodando em ‘0.0.0.0’ na porta ‘8080’
• Habilita o auto-recarregamento quando o código muda via ‘reload=True’

Execute Seu projeto FastAPI

Agora que seu projeto está pronto, vamos executar seu primeiro projeto FastAPI.

Execute o script ‘main.py’ da seguinte forma e seu FastAPI estará rodando em seu sistema.

python3 main.py

Agora abra seu navegador e visite http://SERVERIP:8080/. Se sua instalação for bem-sucedida, você verá a mensagem ‘Hello FastAPI!’. Você também pode acessá-lo via ‘curl’ no terminal.

Por fim, você pode acessar a documentação da API em http://SERVERIP:8080/docs > que é fornecida pelo Swagger UI.

Conectando FastAPI ao MongoDB

Neste guia, você criará uma API básica com FastAPI e MongoDB. Sua API deve ser capaz de CRUD com o servidor de banco de dados MongoDB. Quanto a este passo, você conectará seu projeto ao servidor MongoDB.

Primeiro, execute o comando ‘pip3’ abaixo para instalar o driver MongoDB ‘motor‘ em seu projeto. ‘motor’ fornece acesso à API não bloqueante e baseada em corrotinas ao servidor MongoDB.

pip3 install motor

server/database.py

Após o módulo ‘motor‘ ser instalado, vamos modificar o script ‘server/database.py‘.

Abra o arquivo ‘server/database.py’ com seu editor de texto e insira o seguinte script. Isso será usado para conectar ao servidor MongoDB via módulo ‘motor’.

from motor.motor_asyncio import AsyncIOMotorClient  
  
MONGODB_HOST = "mongodb://localhost:27017"  
  
connection = AsyncIOMotorClient(MONGODB_HOST)  
  
database = connection.items  
item_collection = database.get_collection("item_collection")

• Importa o ‘AsyncIOMotorClient’ do ‘motor.motor_asyncio’
• Cria uma nova constante ‘MONGODB_HOST’ e aponta para o servidor MongoDB ‘mongodb://localhost:27017’
• Conecta ao servidor MongoDB via variável ‘connection’
• Conecta ao banco de dados ‘items’ via variável ‘database’
• Acessa coleções no banco de dados com a variável ‘item_collection’

Criando Modelo de Banco de Dados com pydantic

Nesta seção, você projetará seus dados via ‘pydantic‘, que fornece modelagem para nosso banco de dados MongoDB.

Instale o módulo ‘pydantic’ com o comando ‘pip3’ abaixo. O módulo ‘pydantic‘ é uma biblioteca de validação de dados que permite criar esquemas de banco de dados via modelo.

pip3 install pydantic

Agora abra o arquivo ‘server/models/itemModels.py‘ com seu editor de texto e copie o seguinte script.

from pydantic import BaseModel, Field  
from typing import Optional  
  
class Item(BaseModel):  
 name: str  
 category: str  
 stocks: int  
 price: int = Field(gt=0)  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "Company Smart Watch",  
 "category": "smartwatch",  
 "stocks": 10,  
 "price": 1000,  
 }  
 }  
  
class ItemUpdate(BaseModel):  
 name: Optional[str] = None  
 category: Optional[str] = None  
 stocks: Optional[int] = None  
 price: Optional[int] = None  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "New Smart watch",  
 "category": "new-smartwatch",  
 "stocks": 5,  
 "price": 500,  
 }  
 }

• Importa ‘BaseModel’ e ‘Field’ dos módulos ‘pydantic’
• Importa o módulo ‘Optional’ de ‘typing’
• Cria o seguinte esquema de banco de dados ‘Item’ para o FastAPI: - ‘name’ e ‘category’ com tipo string
• ‘stocks’ e ‘price’ com categoria inteira
• ‘price’ deve ser maior que 0
• Estende o modelo de dados via classe ‘Config’ fornecendo um exemplo de dados que o usuário pode incluir na solicitação
• Cria o seguinte esquema ‘ItemUpdate’ com cada campo como opcional.

Adicionando Operações CRUD

Neste ponto, você criou uma conexão com o servidor MongoDB e criou um esquema de banco de dados via ‘pydantic’. Vamos passar a criar as operações CRUD com FastAPI e MongoDB.

Abra o script ‘server/database.py‘ novamente e crie uma nova função ‘item_helper‘ com tipo ‘dict‘. Chame esta função para obter dados detalhados sobre um item.

def item_helper(item) -> dict:  
 return {  
 "id": str(item["_id"]),  
 "name": item["name"],  
 "category": item["category"],  
 "stocks": item["stocks"],  
 "price": item["price"],  
 }

Adicione a seguinte linha para importar o módulo ‘ObjectId‘ de ‘bson.objectid’. O ‘ObjectId‘ é parte do tipo de dados em BSON, que o MongoDB usa para armazenamento de dados e representação de dados JSON.

from bson.objectid import ObjectId

Criar um Item

Primeiro, você criará uma função assíncrona que será capaz de adicionar novos dados ao banco de dados MongoDB.

Crie uma nova função ‘add_item‘ como a seguinte:

# Adicionar um novo item  
async def add_item(item_details: dict) -> dict :  
 item = await item_collection.insert_one(item_details)  
 new_item = await item_collection.find_one({"_id": item.inserted_id})  
 return item_helper(new_item)

• A função ‘add_item’ aceita dados de dicionário dos detalhes do seu item
• Via ‘item_collection’ para acessar documentos no banco de dados e executar a consulta ‘insert_one’ para adicionar um novo item
• O novo item será impresso após a operação bem-sucedida

Recuperar Todos os Itens

Em segundo lugar, você criará uma nova função ‘get_items’ que será capaz de recuperar todos os itens disponíveis em seu banco de dados.

Crie uma nova função ‘get_items‘ para recuperar todos os itens do MongoDB.

# recuperar todos os itens  
async def get_items():  
 items = []  
 async for item in item_collection.find():  
 items.append(item_helper(item))  
 return items

• Você criará uma lista vazia de ‘items’
• Usando a consulta ‘find()’ para pesquisar todos os dados e percorrê-los
• Cada item será adicionado à lista ‘items’ via método ‘append’
• Após o loop ser concluído, seus itens serão exibidos

Recuperar Item Específico com base no id

Em seguida, você criará uma nova função que será capaz de recuperar dados para um seletor específico ‘id’. Isso mostrará dados detalhados sobre itens específicos.

Crie uma nova função ‘get_item‘ para recuperar dados de itens específicos. A função ‘get_item’ aceitará uma string de ‘id’ como seletor e retornará um dicionário.

# recuperar item específico  
async def get_item(id: str) -> dict:  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 return item_helper(item)  
 return "Item Não Encontrado."

• A consulta ‘find_one()’ será usada para recuperar dados com base no ‘id’
• Se um item for encontrado, os detalhes serão exibidos usando o formato de dicionário ‘item_helper’
• Se o item não estiver disponível, a resposta será ‘Item Não Encontrado’

Atualizar um item

Agora você criará uma nova função para atualizar um item. Você também pode atualizar parcialmente seu item via API.

Defina a nova função ‘change_item‘ como os seguintes códigos:

# atualizar item  
async def change_item(id: str, data: dict):  
 if len(data) < 1:  
 return "Por favor, insira seus dados"  
 find_item = await item_collection.find_one({"_id": ObjectId(id)})  
 if find_item:  
 item_update = await item_collection.update_one({"_id": ObjectId(id)}, {"$set": data})  
 if item_update:  
 return True  
 return False

• A função ‘change_item’ aceita dois argumentos, o ‘id’ do item alvo e ‘data’ como os novos dados
• Se os dados estiverem vazios ou ‘< 1’, a operação é encerrada
• Esta função encontrará um item com base no seletor ‘id’
• Se o ‘id’ for encontrado, a ‘item_update’ será executada
• Se a ‘item_update’ for bem-sucedida, retorna ‘True’, ou retorna ‘False’ se não

Deletar um item

Por fim, você criará a função ‘delete_item’ para deletar itens via um seletor específico.

Para deletar um item, criaremos a função ‘delete_item‘ e usaremos o seletor ‘id’ como a seguinte:

# deletar um item  
async def delete_item(id: str):  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 await item_collection.delete_one({"_id": ObjectId(id)})  
 return(f'Item {id} deletado.')  
 return "Item Não Encontrado."

• A consulta ‘find_one()’ buscará um item com base no ‘id’ fornecido
• Se o item for encontrado, ele será deletado via consulta ‘delete_one()’ e retornará ‘Item id deletado’
• Se o item não estiver disponível, a mensagem ‘Item Não Encontrado’ será exibida

Adicionando Rotas para Operações CRUD

Neste ponto, criamos funções assíncronas para a operação CRUD com FastAPI. Agora vamos criar a rota ou endpoint para cada operação.

Edite o arquivo ‘server/routes/item.py‘ usando seu editor preferido.

server/routes/item.py

Primeiro, adicione os módulos ‘APIRouter‘ e ‘Body‘ do ‘fastapi’. Em seguida, adicione o ‘jsonable_encode‘ do ‘fastapi.encoders’.

from fastapi import APIRouter, Body  
from fastapi.encoders import jsonable_encoder

Importe cada função do arquivo ‘database.py’.

from server.database import (  
 add_item,  
 get_items,  
 get_item,  
 change_item,  
 delete_item,  
)

Importe os modelos ‘Item’ e ‘ItemUpdate’ do arquivo ‘itemModels.py’.

from server.models.itemModels import (  
 Item,  
 ItemUpdate,  
)

Chame a classe ‘APIRouter’ via a variável ‘router’.

router = APIRouter()

Rota para Adicionar Novo Item

Agora vamos adicionar uma rota para adicionar novos itens. Neste exemplo, você pode adicionar um novo item via POST para a URL ‘/item‘.

Adicione as seguintes linhas para configurar rotas para adicionar novos itens. Cada POST na URL raiz do item será tratado como inserção de novos dados.

# operação de adicionar novo item  
@router.post("/")  
async def add_item_data(item: Item = Body(...)):  
 item = jsonable_encoder(item)  
 new_item = await add_item(item)  
 return new_item

• A função ‘add_item_data’ aceita o esquema ‘Item’, que fará parte do ‘Body’ em sua solicitação
• Seu item será convertido para o formato de dicionário via ‘jsonable_encoder’
• Insira seus dados de dicionário via função ‘add_item’ (veja database.py) na variável ‘new_item’
• Retorne seus dados inseridos ‘new_item’ como resposta

Rota para Obter Todos os Itens

Adicione o seguinte script para configurar uma rota para recuperar dados. Cada solicitação GET na URL raiz do item recuperará todos os seus dados.

# obter todos os itens disponíveis  
@router.get("/")  
async def get_item_data():  
 items = await get_items()  
 if items:  
 return items  
 return "Nenhum item disponível."

• Crie a função ‘get_item_data’ que executará a função ‘get_item’ do arquivo ‘database.py’
• Se os itens estiverem disponíveis, você obterá a lista de todos os seus itens
• Se não houver item, você receberá a resposta ‘Nenhum item disponível’

Rota para Obter Item Específico

Para obter detalhes de um item específico, usaremos o ‘id’ como seletor. Cada solicitação GET para ‘/id’ retornará um item detalhado do ‘id’ solicitado.

# Mostrar detalhes do item via id  
@router.get("/{id}")  
async def get_item_details(id):  
 item_details = await get_item(id)  
 if item_details:  
 return item_details  
 return "Item não encontrado."

• A função ‘get_item_details’ será criada e passará o ‘id’ da URL
• A função ‘get_item’ (veja database.py) será chamada e também passará o ‘id’ como argumento
• Se o item for encontrado, os detalhes do item serão exibidos
• Se não houver item com aquele ‘id’ específico, você receberá ‘Item não encontrado’

Rota para Atualizar um Item

Copie o seguinte código para definir a rota para o item atualizado:

# Atualizar Item  
@router.put("/{id}")  
async def update_item(id: str, data: ItemUpdate = Body(...)):  
 data = {k: v for k, v in data.dict().items() if v is not None}  
 updated_item = await change_item(id, data)  
 if updated_item:  
 return{f'Sucesso: item {id} atualizado.'}  
 return "Erro"

• A função ‘update_item’ aceitará dois argumentos, ‘id’ como seletor e ‘data’, que é baseada no modelo ‘ItemUpdate’
• Os dados serão verificados na variável ‘data’
• A variável ‘updated_item’ executará a função ‘change_item’ do arquivo ‘database.py’
• Se a atualização for bem-sucedida, você verá a saída ‘Sucesso’

Rota para Deletar um Item

Insira as seguintes linhas para criar a função ‘remove_item’ para deletar itens. Cada operação DELETE em um específico ‘/id’ removerá o item correspondente ao ‘id’.

# deletar item via id  
@router.delete("/{id}")  
async def remove_item(id):  
 item_to_delete = await delete_item(id)  
 if item_to_delete:  
 return item_to_delete  
 return{f'Item {id} Não Disponível.'}

• A função ‘remove_item’ executará ‘delete_item’ e passará o seletor ‘id’
• A operação de exclusão será armazenada e executada via variável ‘item_to_delete’
• Quando um item não estiver disponível, você receberá a mensagem ‘Item id Não Disponível’

server/app.py

Agora que você terminou o arquivo ‘server/routes/item.py’, vamos incluí-lo no ‘server/app.py’.

Abra o arquivo ‘app.py‘ usando seu editor de texto.

Importe o ‘router’ do arquivo ‘server/routes/item.py‘ como ‘ItemRouter’.

from server.routes.item import router as ItemRouter

Insira o ‘ItemRouter’ com o prefixo padrão ‘/item‘. As operações CRUD serão tratadas via URL ‘/item‘.

app.include_router(ItemRouter, tags=["Item"], prefix="/item")

Agora seu endpoint CRUD estará disponível nas seguintes rotas:

• Adicionar novo item: POST para ‘/item‘
• Recuperar todos os itens: GET para ‘/item‘
• Recuperar item específico: GET para ‘/item/id’. O ‘id’ é gerado pelo MongoDB
• Atualizar item: PUT para ‘/item/id’
• Deletar item: DELETE para ‘/item/id’

Testar Operações CRUD

Primeiro, certifique-se de que seu projeto FastAPI esteja em execução, ou você pode executar o script ‘main.py’ como o seguinte:

python3 main.py

Navegue até http://SERVERIP:8080/docs e você verá cada rota que você criou.

Abaixo está um exemplo de adicionar um novo item.

Recupere todos os itens disponíveis via API.

Recupere itens específicos via o seletor ‘id’.

Atualize dados parciais de itens específicos.

Abaixo estão os dados atualizados.

Abaixo está a operação de exclusão via o seletor ‘id’.

Conclusão

Parabéns! Você completou a instalação do FastAPI com MongoDB no Ubuntu 24.04. Você também aprendeu como conectar o FastAPI ao MongoDB usando o módulo ‘motor’, criar modelos de dados via ‘pydantic’, criar operações CRUD com FastAPI e aprender como criar endpoints para sua API.