Cómo instalar Fast API con MongoDB en Ubuntu 24.04

FastAPI es un framework web basado en Python para crear servicios API. Es un framework moderno, rápido y de alto rendimiento que soporta operaciones asíncronas.

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 empezar con esta guía, asegúrate de que tienes lo siguiente:

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

Instalación de MongoDB

Antes de crear un nuevo proyecto FastAPI, vamos a instalar el servidor MongoDB en nuestro sistema.

En primer lugar, 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 añadir 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

Añade el repositorio MongoDB utilizando el comando que aparece a continuación.

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

Una vez añadido el repositorio, ejecuta el siguiente comando para actualizar tu índice de paquetes e instalar el servidor MongoDB. Introduce «Y» para confirmar la instalación.

sudo apt update && sudo apt install mongodb-org

Una vez finalizada la instalación, inicia y habilita el servicio MongoDB ‘mongod’ y, a continuación, comprueba el estado de tu servicio MongoDB para asegurarte de que se está ejecutando.

sudo systemctl enable --now mongod
sudo systemctl status mongod

A continuación puedes ver que el servidor MongoDB se está ejecutando.

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

mongosh

Configurar Python y el entorno virtual

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

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

sudo apt install python3 python3-pip python3-venv

Una vez finalizada la instalación, inicia sesión con tu usuario.

su - username

Ahora crea un nuevo directorio‘~/app‘ y entra en é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 actívalo. De este modo, tu intérprete de comandos será como ‘(venv) usuario@nombrehost‘.

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 la sesión en «venv» utilizando el siguiente comando.

deactivate

Creación del proyecto FastAPI

Ahora que has creado y activado tu entorno virtual Python, vamos a instalar FastAPI y a crear la estructura del proyecto.

Ejecuta el comando ‘pip3’ para instalar los paquetes ‘fastapi’ y ‘uvicorn’.

pip3 install fastapi uvicorn

• El ‘fastapi’ es el principal framework web FastAPI para construir APIs en Python
• El ‘uvicorn’ es la implementación del servidor web ASGI (Asynchronous Server Gateway Interface) en Python.

Una vez finalizada la instalación, 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 la estructura de nuestro proyecto FastAPI.

servidor/app.py

Ahora que tu proyecto está listo, vamos a modificar 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": "Hello FastAPI!"}

• Importa el módulo FastAPI a tu proyecto y vincúlalo a la variable ‘app
• Crea una nueva función ‘raíz’ que devuelva el mensaje ‘¡Hola FastAPI!
• La función ‘root’ responde al método GET de la URL raíz
• async’ marca tu función como función asíncrona y puede utilizar ‘await’ dentro de su cuerpo cuando sea llamada

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)

• Importar el módulo «uvicorn
• Cuando se ejecute el script ‘main.py’, cargará el módulo ‘app’ o FastAPI dentro del ‘server/app.py’
• FastAPI se ejecutará en «0.0.0.0» con el puerto «8080».
• Activa la auto-recarga cuando cambie el código mediante ‘reload=True’

Ejecuta tu proyecto FastAPI

Ahora que tu proyecto está listo, vamos a ejecutar tu primer proyecto FastAPI.

Ejecuta el script ‘main.py’ como se indica a continuación y tu FastAPI estará funcionando en tu sistema.

python3 main.py

Ahora abre tu navegador web y visita http://SERVERIP:8080/. Si la instalación se ha realizado correctamente, verás el mensaje ‘¡Hola FastAPI! También puedes acceder a ella mediante ‘curl’ desde el terminal.

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

Conectar FastAPI a MongoDB

En esta guía, crearás una API básica con FastAPI y MongoDB. Tu API debe ser capaz de hacer CRUD con el servidor de base de datos MongoDB. En este paso, conectarás tu proyecto al servidor MongoDB.

En primer lugar, ejecuta el siguiente comando ‘pip3’ para instalar el controlador‘motor’ de MongoDB en tu proyecto. ‘motor’ proporciona un acceso API no bloqueante y basado en coroutinas al servidor MongoDB.

pip3 install motor

servidor/base de datos.py

Una vez instalado el módulo«motor«, vamos a modificar el script«server/database.py«.

Abre el archivo ‘server/database.py’ con tu editor de texto e introduce el siguiente script. Se utilizará para conectar con el 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 el ‘AsyncIOMotorClient’ del módulo ‘motor.motor_asyncio’.
• Crea una nueva constante «MONGODB_HOST» y apunta al servidor MongoDB «mongodb://localhost:27017».
• Conéctate al servidor MongoDB mediante la variable ‘connection
• Conéctate a la base de datos ‘items’ mediante la variable ‘database
• Accede a las colecciones de la base de datos con la variable ‘item_collection

Crear un modelo de base de datos con pydantic

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

Instala el módulo ‘pydantic’ con el siguiente comando ‘pip3’. El módulo‘pydantic‘ es una biblioteca de validación de datos que te permite crear esquemas de bases de datos mediante 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": "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 los módulos ‘BaseModel’ y ‘Field’ de ‘pydantic’.
• Importa el módulo «Opcional» de «typing».
• Crea el siguiente esquema de base de datos «Elemento» para la FastAPI:
  • ‘nombre’ y ‘categoría’ con tipo cadena
  • ‘existencias’ y ‘precio’ con categoría entero
  • ‘precio’ debe ser mayor que 0
• Amplía el modelo de datos mediante la clase ‘Config’ proporcionando un ejemplo de datos que el usuario puede incluir en la solicitud
• Crea el siguiente esquema ‘ItemUpdate’ con todos los campos como opcionales.

Añadir operaciones CRUD

Llegados a este punto, ya has creado una conexión con el servidor MongoDB y creado el esquema de la base de datos mediante ‘pydantic’. Pasemos a crear las operaciones CRUD con FastAPI y MongoDB.

Abre de nuevo 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 elemento.

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

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

from bson.objectid import ObjectId

Crear un elemento

En primer lugar, crearás una función asíncrona que podrá añadir nuevos datos a la base de datos MongoDB.

Crea una nueva función ‘add_item‘ como la siguiente:

# Add a new 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)

• La función ‘add_item’ acepta datos de diccionario de los detalles de tu artículo
• A través de ‘item_collection’ accede a los documentos de la base de datos y ejecuta la consulta ‘insert_one’ para añadir un nuevo elemento
• El nuevo artículo se imprimirá si la operación se realiza correctamente

Recuperar todos los artículos

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

Crea una nueva función‘get_items‘ para recuperar todos los elementos de MongoDB.

# retrieve all items
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 ‘ítems’.
• Utilizarás la consulta ‘find()’ para buscar todos los datos y hacer un bucle a través de ellos
• Cada elemento se añadirá a la lista ‘items’ mediante el método ‘append
• Una vez completado el bucle, se mostrarán tus elementos

Recuperar un elemento específico basado en el id

A continuación, crearás una nueva función que podrá recuperar datos para el 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 elementos específicos. La función ‘get_item’ aceptará una cadena de ‘id’ como selector y devolverá un diccionario.

# retrieve specific item
async def get_item(id: str) -> dict:
 item = await item_collection.find_one({"_id": ObjectId(id)})
 if item:
 return item_helper(item)
 return "Item Not Found."

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

Actualizar un artículo

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

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

# update item
async def change_item(id: str, data: dict):
 if len(data) < 1:
 return "Please input your data"
 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 elemento de destino y ‘data’ como los nuevos datos
• Si los datos están vacíos o son ‘< 1’, la operación finaliza
• Esta función buscará un elemento basándose en el selector «id
• Si se encuentra el ‘id’, se ejecutará ‘item_update
• Si el ‘item_update’ tiene éxito, devuelve ‘True’, o devuelve ‘False’ en caso contrario

Eliminar un elemento

Por último, crearemos la función ‘eliminar_elemento’ para eliminar elementos mediante un selector específico.

Para eliminar un elemento, crearemos la función«eliminar_elemento» y utilizaremos el selector «id» como se indica a continuación:

# delete an 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} deleted.')
 return "Item Not Found."

• La consulta ‘find_one()’ buscará un elemento basándose en el ‘id’ proporcionado
• Si se encuentra el elemento, se eliminará mediante la consulta ‘delete_one()’ y devolverá el ‘Id de elemento eliminado’
• Si el elemento no está disponible, se mostrará el mensaje «Elemento no encontrado».

Añadir rutas para operaciones CRUD

Llegados a este punto, hemos creado funciones asíncronas para operaciones CRUD con FastAPI. Ahora vamos a crear la ruta o endpoint para cada operación.

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

servidor/rutas/elemento.py

Primero, añade los módulos ‘APIRouter‘ y‘Body’ de ‘fastapi’. A continuación, añade el módulo‘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 añadir un nuevo artículo

Ahora vamos a añadir una ruta para añadir nuevos artículos. En este ejemplo, puedes añadir un nuevo elemento mediante POST a la URL‘/item‘.

Añade las siguientes líneas para configurar las rutas para añadir nuevos elementos. Cada POST a la URL raíz del elemento se tratará como inserción de nuevos datos.

# add new item operation
@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 ‘Artículo’, que formará parte del ‘Cuerpo’ de tu petición
• Tu ítem se convertirá a formato diccionario mediante ‘jsonable_encoder’
• Inserta tus datos de diccionario mediante la función ‘add_item’ (ver base_de_datos.py) sobre la variable ‘new_item
• Devuelve tus datos insertados ‘nuevo_elemento’ como respuesta

Ruta para obtener todos los elementos

Añade el siguiente script para configurar una ruta para recuperar datos. Cada petición GET a la URL raíz del elemento recuperará todos tus datos.

# get all available items
@router.get("/")
async def get_item_data():
 items = await get_items()
 if items:
 return items
 return "No available item."

• 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 ningún artículo, obtendrás la respuesta «Ningún artículo disponible

Ruta para obtener un artículo concreto

Para obtener los detalles de un artículo concreto, utilizaremos el «id» como selector. Cada solicitud GET al ‘/id’ devolverá un artículo detallado del ‘id’ solicitado.

# Show details of the item via the id
@router.get("/{id}")
async def get_item_details(id):
 item_details = await get_item(id)
 if item_details:
 return item_details
 return "Item not found."

• Se creará la función ‘get_item_details’ y se le pasará el ‘id’ de la URL
• Se llamará a la función ‘get_item’ (ver base de datos.py) y también se le pasará el ‘id’ como argumento
• Si se encuentra el artículo, se mostrarán los detalles del mismo
• Si no hay ningún 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 elemento actualizado:

# Update 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'Success: item {id} updated.'}
 return "Error"

• La función ‘update_item’ tomará dos argumentos, ‘id’ como selector y los ‘datos’, que se basan en el modelo ‘ItemUpdate
• Los datos se comprobarán en la variable ‘datos
• El ‘elemento_actualizado’ ejecutará la función ‘cambiar_elemento’ del archivo ‘base_de_datos.py
• Si la actualización tiene éxito, verás la salida ‘Éxito

Ruta para eliminar un elemento

Inserta las siguientes líneas para crear la función ‘eliminar_elemento’ para borrar elementos. Cada operación DELETE a un ‘/id’ específico, eliminará el elemento coincidente con el ‘id’.

# delete 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} Not Available.'}

• La función ‘eliminar_elemento’ ejecutará ‘eliminar_elemento’ 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 elemento no esté disponible, obtendrás el retorno ‘Id de elemento no disponible’

servidor/app.py

Ahora que has terminado el archivo ‘servidor/rutas/artículo.py’, vamos a incluirlo en el archivo ‘servidor/app.py’.

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

Importa el ‘enrutador’ del archivo‘servidor/rutas/elemento.py‘ como un‘ItemRouter‘.

from server.routes.item import router as ItemRouter

Inserta el ‘ItemRouter’ con el prefijo por defecto‘/item‘. Las operaciones CRUD se gestionará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:

• Añadir nuevo elemento: POST a‘/artículo
• Recuperar todos los elementos: GET a‘/artículo
• Recuperar un elemento específico: GET a ‘/artículo/id’. El ‘id’ lo genera MongoDB
• Actualizar un elemento: PUT a ‘/artículo/id’
• Eliminar elemento: DELETE a ‘/artículo/id’

Prueba las operaciones CRUD

En primer lugar, asegúrate de que tu proyecto FastAPI se está ejecutando, o puedes ejecutar el script ‘main.py’ como el siguiente:

python3 main.py

Navega por http://SERVERIP:8080/docs y verás cada ruta que hayas creado.

A continuación se muestra un ejemplo de adición de un nuevo elemento.

Recuperar todos los elementos disponibles a través de la API.

Recuperar elementos específicos mediante el selector «id».

Actualizar datos parciales de elementos específicos.

A continuación se muestran los datos actualizados.

A continuación se muestra la operación de eliminación mediante el selector «id».

Conclusión

¡Enhorabuena! Has completado la instalación de FastAPI con MongoDB en Ubuntu 24.04. También has aprendido a conectar FastAPI a MongoDB mediante el módulo ‘motor’, a crear modelos de datos mediante ‘pydantic’, a crear operaciones CRUD con FastAPI y a crear endpoints para tu API.