Cómo instalar Fast API con MongoDB en Ubuntu 24.04

FastAPI es un marco web basado en Python para crear servicios API. Es un marco moderno, rápido y de alto rendimiento que admite operaciones asincrónicas.

En este tutorial, aprenderás a instalar FastAPI con MongoDB en Ubuntu 24.04. También aprenderás a crear tu primera API con operaciones CRUD utilizando FastAPI y la base de datos MongoDB.

Requisitos previos

Para comenzar con esta guía, asegúrate de tener lo siguiente:

• Un sistema Ubuntu 24.04
• Un usuario no root con privilegios de administrador

Instalando MongoDB

Antes de crear un nuevo proyecto FastAPI, instalemos el servidor MongoDB en nuestro sistema.

Primero, ejecuta el siguiente comando para actualizar tu índice de paquetes e instalar ‘gnupg’ y ‘curl’ en tu sistema.

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

Ejecuta el siguiente comando para agregar la clave GPG para el 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

Agrega el repositorio de MongoDB usando el siguiente comando.

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

Después de agregar el repositorio, ejecuta el siguiente comando para actualizar tu índice de paquetes e instalar el servidor MongoDB. Ingresa ‘Y’ para confirmar la instalación.

sudo apt update && sudo apt install mongodb-org

Cuando la instalación haya finalizado, inicia y habilita el servicio de MongoDB ‘mongod’, y luego verifica el estado de tu servicio MongoDB para asegurarte de que esté en funcionamiento.

sudo systemctl enable --now mongod  
sudo systemctl status mongod

Puedes ver a continuación que el servidor MongoDB está en funcionamiento.

Además de eso, puedes iniciar sesión en el servidor MongoDB con el comando ‘mongosh’ a continuación. Para salir, presiona Ctrl+d.

mongosh

Configurando Python y Entorno Virtual

Con MongoDB instalado, instalarás paquetes de Python y configurarás el directorio del proyecto y el entorno virtual.

Instala Python, Pip y los módulos Venv con el siguiente comando. Ingresa ‘Y’ para confirmar la instalación.

sudo apt install python3 python3-pip python3-venv

Una vez que la instalación haya finalizado, inicia sesión en tu usuario.

su - username

Ahora crea un nuevo directorio ‘~/app‘ y muévete a él. Este directorio se utilizará para almacenar tu proyecto FastAPI.

mkdir -p ~/app; cd ~/app

Ejecuta el siguiente comando para crear un nuevo entorno virtual ‘venv‘ y activarlo. Con esto, tu indicador de shell se verá como ‘(venv) user@hostname‘.

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

A partir de aquí, tu entorno de trabajo debe estar en el entorno virtual ‘venv’. Puedes cerrar sesión de ‘venv’ usando el siguiente comando.

deactivate

Creando Proyecto FastAPI

Ahora que has creado y activado tu entorno virtual de Python, instalemos FastAPI y creemos la estructura del proyecto.

Con el comando ‘pip3’, ejecútalo para instalar los paquetes ‘fastapi’ y ‘uvicorn’.

pip3 install fastapi uvicorn

• El ‘fastapi’ es el marco web principal de FastAPI para construir APIs en Python
• El ‘uvicorn’ es la implementación del servidor web ASGI (Interfaz de puerta de enlace de servidor asincrónico) en Python.

Después de que la instalación haya finalizado, crea nuevos archivos y directorios con el siguiente comando.

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

A continuación se muestra nuestra estructura del proyecto FastAPI.

server/app.py

Ahora que tu proyecto está listo, modifiquemos el archivo ‘server/app.py’, que es la aplicación principal de tu proyecto FastAPI.

Abre el archivo ‘app.py‘ con tu editor de texto y copia el siguiente script.

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

• Importa el módulo FastAPI a tu proyecto y asígnalo a la variable ‘app’
• Crea una nueva función ‘root’ que devuelve ‘¡Hola FastAPI!’
• La función ‘root’ responde al método GET en la URL raíz
• ‘async’ marca tu función como una función asincrónica y puede usar ‘await’ dentro de su cuerpo cuando se llama

main.py

En esta sección, modificaremos el archivo ‘main.py’ que se utilizará para ejecutar tu proyecto FastAPI a través de ‘uvicorn’ (servidor web ASGI en Python).

Ahora abre y modifica el script ‘main.py‘ e inserta el siguiente código.

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

• Importa el módulo ‘uvicorn’
• Cuando se ejecute el script ‘main.py’, cargará el módulo ‘app’ o FastAPI dentro de ‘server/app.py’
• FastAPI se ejecutará en ‘0.0.0.0’ con el puerto ‘8080’
• Habilita la recarga automática cuando el código cambia a través de ‘reload=True’

Ejecuta tu proyecto FastAPI

Ahora que tu proyecto está listo, ejecutemos tu primer proyecto FastAPI.

Ejecuta el script ‘main.py’ como sigue y tu FastAPI se ejecutará en tu sistema.

python3 main.py

Ahora abre tu navegador web y visita http://SERVERIP:8080/. Si tu instalación es exitosa, verás el mensaje ‘¡Hola FastAPI!’. También puedes acceder a él a través de ‘curl’ desde la terminal.

Por último, puedes acceder a la documentación de la API en http://SERVERIP:8080/docs > que es proporcionada por Swagger UI.

Conectando FastAPI a MongoDB

En esta guía, crearás una API básica con FastAPI y MongoDB. Tu API debería poder realizar operaciones CRUD con el servidor de base de datos MongoDB. En este paso, conectarás tu proyecto al servidor MongoDB.

Primero, ejecuta el comando ‘pip3’ a continuación para instalar el controlador ‘ motor ‘ de MongoDB en tu proyecto. ‘motor’ proporciona acceso a la API no bloqueante y basado en corutinas al servidor MongoDB.

pip3 install motor

server/database.py

Después de que el módulo ‘ motor ‘ esté instalado, modifiquemos el script ‘ server/database.py ‘.

Abre el archivo ‘server/database.py’ con tu editor de texto e ingresa el siguiente script. Esto se utilizará para conectarse al servidor MongoDB a través del 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 ‘AsyncIOMotorClient’ de ‘motor.motor_asyncio’
• Crea una nueva constante ‘MONGODB_HOST’ y apunta al servidor MongoDB ‘mongodb://localhost:27017’
• Conéctate al servidor MongoDB a través de la variable ‘connection’
• Conéctate a la base de datos ‘items’ a través de la variable ‘database’
• Accede a las colecciones en la base de datos con la variable ‘item_collection’

Creando Modelo de Base de Datos con pydantic

En esta sección, diseñarás tus datos a través de ‘ pydantic ‘, que proporciona modelado para nuestra base de datos MongoDB.

Instala el módulo ‘pydantic’ con el comando ‘pip3’ a continuación. El módulo ‘ pydantic ‘ es una biblioteca de validación de datos que te permite crear esquemas de base de datos a través de modelos.

pip3 install pydantic

Ahora abre el archivo ‘server/models/itemModels.py‘ con tu editor de texto y copia el siguiente 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": "Reloj inteligente de la empresa",  
 "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": "Nuevo reloj inteligente",  
 "category": "nuevo-smartwatch",  
 "stocks": 5,  
 "price": 500,  
 }  
 }

• Importa los módulos ‘BaseModel’ y ‘Field’ de ‘pydantic’
• Importa el módulo ‘Optional’ de ‘typing’
• Crea el siguiente esquema de base de datos ‘Item’ para FastAPI: - ‘name’ y ‘category’ con tipo cadena
• ‘stocks’ y ‘price’ con categoría entera
• ‘price’ debe ser mayor que 0
• Extiende el modelo de datos a través de la clase ‘Config’ proporcionando un ejemplo de datos que el usuario puede incluir en la solicitud
• Crea el siguiente esquema ‘ItemUpdate’ con cada campo como opcional.

Agregando Operaciones CRUD

En este punto, has creado una conexión al servidor MongoDB y creado un esquema de base de datos a través de ‘pydantic’. Pasemos a crear las operaciones CRUD con FastAPI y MongoDB.

Abre nuevamente el script ‘ server/database.py ‘ y crea una nueva función ‘ item_helper ‘ con tipo ‘ dict ‘. Llama a esta función para obtener datos detallados sobre un artículo.

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

Agrega la siguiente línea para importar el módulo ‘ ObjectId ‘ de ‘bson.objectid’. El ‘ ObjectId ‘ es parte del tipo de datos en BSON, que MongoDB utiliza para el almacenamiento de datos y la representación de datos JSON.

from bson.objectid import ObjectId

Crear un Artículo

Primero, crearás una función asincrónica que podrá agregar nuevos datos a la base de datos MongoDB.

Crea una nueva función ‘a dd_item ‘ como la siguiente:

# Agregar un nuevo artículo  
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)

• La función ‘add_item’ acepta datos de diccionario de los detalles de tu artículo
• A través de ‘item_collection’ para acceder a documentos en la base de datos y ejecutar la consulta ‘insert_one’ para agregar un nuevo artículo
• El nuevo artículo se imprimirá al finalizar la operación con éxito

Recuperar Todos los Artículos

En segundo lugar, crearás una nueva función ‘get_items’ que podrá recuperar todos los artículos disponibles en tu base de datos.

Crea una nueva función ‘ get_items ‘ para recuperar todos los artículos de MongoDB.

# recuperar todos los artículos  
async def get_items():  
 items = []  
 async for item in item_collection.find():  
 items.append(item_helper(item))  
 return items

• Crearás una lista vacía de ‘items’
• Usando la consulta ‘find()’ para buscar todos los datos y recorrerlo
• Cada artículo se agregará a la lista ‘items’ a través del método ‘append’
• Después de que el bucle esté completo, tus artículos se mostrarán

Recuperar Artículo Específico basado en id

A continuación, crearás una nueva función que podrá recuperar datos para un selector específico ‘id’. Esto te mostrará datos detallados sobre artículos específicos.

Crea una nueva función ‘ get_item ‘ para recuperar datos de artículos específicos. La función ‘get_item’ aceptará una cadena de ‘id’ como selector y devolverá un diccionario.

# recuperar artículo 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 "Artículo no encontrado."

• La consulta ‘find_one()’ se utilizará para recuperar datos basados en el ‘id’
• Si se encuentra un artículo, los detalles se mostrarán utilizando el formato de diccionario ‘item_helper’
• Si el artículo no está disponible, la respuesta es ‘Artículo no encontrado’

Actualizar un artículo

Ahora crearás una nueva función para actualizar un artículo. También puedes actualizar parcialmente tu artículo a través de la API.

Define la nueva función ‘ change_item ‘ como los siguientes códigos:

# actualizar artículo  
async def change_item(id: str, data: dict):  
 if len(data) < 1:  
 return "Por favor ingresa tus datos"  
 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

• La función ‘change_item’ toma dos argumentos, el ‘id’ del artículo objetivo y ‘data’ como los nuevos datos
• Si los datos están vacíos o ‘< 1’, la operación se termina
• Esta función encontrará un artículo basado en el selector ‘id’
• Si se encuentra el ‘id’, se ejecutará ‘item_update’
• Si ‘item_update’ tiene éxito, devuelve ‘True’, o devuelve ‘False’ si no

Eliminar un artículo

Por último, crearás la función ‘delete_item’ para eliminar artículos a través de un selector específico.

Para eliminar un artículo, crearemos la función ‘ delete_item ‘ y usaremos el selector ‘id’ como sigue:

# eliminar un artículo  
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'Artículo {id} eliminado.')  
 return "Artículo no encontrado."

• La consulta ‘find_one()’ buscará un artículo basado en el ‘id’ proporcionado
• Si se encuentra el artículo, se eliminará a través de la consulta ‘delete_one()’ y devolverá ‘Artículo id eliminado’
• Si el artículo no está disponible, se mostrará ‘Artículo no encontrado’

Agregando Rutas para Operaciones CRUD

En este punto, hemos creado funciones asincrónicas para la operación CRUD con FastAPI. Ahora vamos a crear la ruta o el punto final para cada operación.

Edita el archivo ‘ server/routes/item.py ‘ usando tu editor preferido.

server/routes/item.py

Primero, agrega los módulos ‘ APIRouter ‘ y ‘ Body ‘ de ‘fastapi’. Luego, agrega el ‘ jsonable_encode ‘ de ‘fastapi.encoders’.

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

Importa cada función del archivo ‘database.py’.

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

Importa los modelos ‘Item’ y ‘ItemUpdate’ del archivo ‘itemModels.py’.

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

Llama a la clase ‘APIRouter’ a través de la variable ‘router’.

router = APIRouter()

Ruta para Agregar Nuevo Artículo

Ahora agreguemos una ruta para agregar nuevos artículos. En este ejemplo, puedes agregar un nuevo artículo a través de POST a la URL ‘ /item ‘.

Agrega las siguientes líneas para configurar rutas para agregar nuevos artículos. Cada POST a la URL raíz del artículo se tratará como insertar nuevos datos.

# operación para agregar nuevo artículo  
@router.post("/")  
async def add_item_data(item: Item = Body(...)):  
 item = jsonable_encoder(item)  
 new_item = await add_item(item)  
 return new_item

• La función ‘add_item_data’ acepta el esquema ‘Item’, que será parte del ‘Body’ en tu solicitud
• Tu artículo se convertirá a formato de diccionario a través de ‘jsonable_encoder’
• Inserta tus datos de diccionario a través de la función ‘add_item’ (ver database.py) en la variable ‘new_item’
• Devuelve tus datos insertados ‘new_item’ como respuesta

Ruta para Obtener Todos los Artículos

Agrega el siguiente script para configurar una ruta para recuperar datos. Cada solicitud GET a la URL raíz del artículo recuperará todos tus datos.

# obtener todos los artículos disponibles  
@router.get("/")  
async def get_item_data():  
 items = await get_items()  
 if items:  
 return items  
 return "No hay artículos disponibles."

• Crea la función ‘get_item_data’ que ejecutará la función ‘get_item’ del archivo ‘database.py’
• Si hay artículos disponibles, obtendrás la lista de todos tus artículos
• Si no hay artículo, obtendrás la respuesta ‘No hay artículos disponibles’

Ruta para Obtener Artículo Específico

Para obtener detalles de un artículo específico, utilizaremos el ‘id’ como selector. Cada solicitud GET a ‘/id’ devolverá un artículo detallado del ‘id’ solicitado.

# Mostrar detalles del artículo a través del id  
@router.get("/{id}")  
async def get_item_details(id):  
 item_details = await get_item(id)  
 if item_details:  
 return item_details  
 return "Artículo no encontrado."

• La función ‘get_item_details’ se creará y pasará el ‘id’ desde la URL
• La función ‘get_item’ (ver database.py) se llamará y también pasará el ‘id’ como argumento
• Si se encuentra el artículo, se mostrarán los detalles del artículo
• Si no hay artículo con ese ‘id’ específico, obtendrás ‘Artículo no encontrado’

Ruta para Actualizar un Artículo

Copia el siguiente código para establecer la ruta para el artículo actualizado:

# Actualizar Artículo  
@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'Éxito: artículo {id} actualizado.'}  
 return "Error"

• La función ‘update_item’ tomará dos argumentos, ‘id’ como el selector y ‘data’, que se basa en el modelo ‘ItemUpdate’
• Los datos se verificarán en la variable ‘data’
• La variable ‘updated_item’ ejecutará la función ‘change_item’ del archivo ‘database.py’
• Si la actualización tiene éxito, verás la salida ‘Éxito’

Ruta para Eliminar un Artículo

Inserta las siguientes líneas para crear la función ‘remove_item’ para eliminar artículos. Cada operación DELETE a un ‘/id’ específico eliminará el artículo que coincida con el ‘id’.

# eliminar artículo a través del 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'Artículo {id} no disponible.'}

• La función ‘remove_item’ ejecutará ‘delete_item’ y pasará el selector ‘id’
• La operación de eliminación se almacenará y ejecutará a través de la variable ‘item_to_delete’
• Cuando un artículo no esté disponible, obtendrás la respuesta ‘Artículo id no disponible’

server/app.py

Ahora que has terminado el archivo ‘server/routes/item.py’, incluyámoslo en el ‘server/app.py’.

Abre el archivo ‘app.py‘ usando tu editor de texto.

Importa el ‘router’ del archivo ‘server/routes/item.py‘ como ‘ ItemRouter ‘.

from server.routes.item import router as ItemRouter

Inserta el ‘ItemRouter’ con el prefijo predeterminado ‘ /item ‘. Las operaciones CRUD se manejarán a través de la URL ‘ /item ‘.

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

Ahora tu punto final CRUD estará disponible en lo siguiente:

• Agregar nuevo artículo: POST a ‘ /item ‘
• Recuperar todos los artículos: GET a ‘ /item ‘
• Recuperar artículo específico: GET a ‘/item/id’. El ‘id’ es generado por MongoDB
• Actualizar artículo: PUT a ‘/item/id’
• Eliminar artículo: DELETE a ‘/item/id’

Probar Operaciones CRUD

Primero, asegúrate de que tu proyecto FastAPI esté en funcionamiento, o puedes ejecutar el script ‘main.py’ como sigue:

python3 main.py

Navega a través de http://SERVERIP:8080/docs y verás cada ruta que has creado.

A continuación se muestra un ejemplo de agregar un nuevo artículo.

Recupera todos los artículos disponibles a través de la API.

Recupera artículos específicos a través del selector ‘id’.

Actualiza datos parciales de artículos específicos.

A continuación se muestran los datos actualizados.

A continuación se muestra la operación de eliminación a través del selector ‘id’.

Conclusión

¡Felicidades! Has completado la instalación de FastAPI con MongoDB en Ubuntu 24.04. También has aprendido cómo conectar FastAPI a MongoDB utilizando el módulo ‘motor’, crear modelos de datos a través de ‘pydantic’, crear operaciones CRUD con FastAPI y aprender cómo crear puntos finales para tu API.