Как установить Fast API с MongoDB на Ubuntu 24.04

FastAPI — это веб-фреймворк на Python для создания API-сервисов. Это современный, быстрый, высокопроизводительный фреймворк, поддерживающий асинхронные операции.

В этом руководстве вы узнаете, как установить FastAPI с MongoDB на Ubuntu 24.04. Вы также научитесь создавать свой первый API с операциями CRUD, используя FastAPI и базу данных MongoDB.

Предварительные требования

Для начала убедитесь, что у вас есть следующее:

• Система Ubuntu 24.04
• Пользователь без прав root с правами администратора

Установка MongoDB

Перед созданием нового проекта FastAPI давайте установим сервер MongoDB на нашу систему.

Сначала выполните команду ниже, чтобы обновить индекс пакетов и установить ‘gnupg’ и ‘curl’ на вашу систему.

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

Выполните команду ниже, чтобы добавить GPG-ключ для сервера 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

Добавьте репозиторий MongoDB, используя команду ниже.

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

После добавления репозитория выполните следующую команду, чтобы обновить индекс пакетов и установить сервер MongoDB. Введите ‘Y’, чтобы подтвердить установку.

sudo apt update && sudo apt install mongodb-org

Когда установка завершится, запустите и включите службу MongoDB ‘mongod’, а затем проверьте статус службы MongoDB, чтобы убедиться, что она работает.

sudo systemctl enable --now mongod  
sudo systemctl status mongod

Вы можете увидеть ниже, что сервер MongoDB работает.

Кроме того, вы можете войти в сервер MongoDB с помощью команды ‘mongosh’ ниже. Чтобы выйти, нажмите Ctrl+d.

mongosh

Настройка Python и виртуального окружения

С установленным MongoDB вы установите пакеты Python и настроите каталог проекта и виртуальное окружение.

Установите модули Python, Pip и Venv с помощью следующей команды. Введите ‘Y’, чтобы подтвердить установку.

sudo apt install python3 python3-pip python3-venv

После завершения установки войдите в свою учетную запись.

su - username

Теперь создайте новый каталог ‘~/app‘ и перейдите в него. Этот каталог будет использоваться для хранения вашего проекта FastAPI.

mkdir -p ~/app; cd ~/app

Выполните команду ниже, чтобы создать новое виртуальное окружение ‘venv‘ и активировать его. Ваша командная строка станет выглядеть как ‘(venv) user@hostname‘.

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

С этого момента ваша рабочая среда должна находиться в виртуальном окружении ‘venv’. Вы можете выйти из ‘venv’, используя команду ниже.

deactivate

Создание проекта FastAPI

Теперь, когда вы создали и активировали свое виртуальное окружение Python, давайте установим FastAPI и создадим структуру проекта.

С помощью команды ‘pip3’ выполните установку пакетов ‘fastapi’ и ‘uvicorn’.

pip3 install fastapi uvicorn

• ‘fastapi’ — это основной веб-фреймворк FastAPI для создания API на Python
• ‘uvicorn’ — это реализация веб-сервера ASGI (Asynchronous Server Gateway Interface) на Python.

После завершения установки создайте новые файлы и каталоги с помощью следующей команды.

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

Ниже представлена структура нашего проекта FastAPI.

server/app.py

Теперь, когда ваш проект готов, давайте изменим файл ‘server/app.py’, который является основным приложением вашего проекта FastAPI.

Откройте файл ‘app.py‘ с помощью текстового редактора и скопируйте следующий скрипт.

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

• Импортируйте модуль FastAPI в ваш проект и свяжите его с переменной ‘app’
• Создайте новую функцию ‘root’, которая возвращает ‘Hello FastAPI!’
• Функция ‘root’ отвечает на метод GET по корневому URL
• ‘async’ помечает вашу функцию как асинхронную и может использовать ‘await’ внутри своего тела при вызове

main.py

В этом разделе мы изменим файл ‘main.py’, который будет использоваться для запуска вашего проекта FastAPI через ‘uvicorn’ (ASGI веб-сервер на Python).

Теперь откройте и измените скрипт ‘main.py‘ и вставьте следующий код.

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

• Импортируйте модуль ‘uvicorn’
• Когда скрипт ‘main.py’ выполняется, он загрузит модуль ‘app’ или FastAPI в ‘server/app.py’
• FastAPI будет работать на ‘0.0.0.0’ с портом ‘8080’
• Включите автоматическую перезагрузку при изменении кода через ‘reload=True’

Запустите ваш проект FastAPI

Теперь, когда ваш проект готов, давайте запустим ваш первый проект FastAPI.

Выполните скрипт ‘main.py’ следующим образом, и ваш FastAPI будет работать на вашей системе.

python3 main.py

Теперь откройте веб-браузер и перейдите по адресу http://SERVERIP:8080/. Если ваша установка успешна, вы увидите сообщение ‘Hello FastAPI!’. Вы также можете получить к нему доступ через ‘curl’ из терминала.

Наконец, вы можете получить доступ к документации API по адресу http://SERVERIP:8080/docs >, которая предоставляется Swagger UI.

Подключение FastAPI к MongoDB

В этом руководстве вы создадите базовый API с FastAPI и MongoDB. Ваш API должен уметь выполнять операции CRUD с сервером базы данных MongoDB. На этом этапе вы подключите ваш проект к серверу MongoDB.

Сначала выполните команду ‘pip3’ ниже, чтобы установить драйвер ‘ motor ‘ MongoDB в ваш проект. ‘motor’ предоставляет неблокирующий и основанный на корутинах доступ к API сервера MongoDB.

pip3 install motor

server/database.py

После установки модуля ‘ motor ‘ давайте изменим скрипт ‘ server/database.py ‘.

Откройте файл ‘server/database.py’ с помощью текстового редактора и введите следующий скрипт. Это будет использоваться для подключения к серверу MongoDB через модуль ‘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")

• Импортируйте ‘AsyncIOMotorClient’ из ‘motor.motor_asyncio’
• Создайте новую константу ‘MONGODB_HOST’ и укажите сервер MongoDB ‘mongodb://localhost:27017’
• Подключитесь к серверу MongoDB через переменную ‘connection’
• Подключитесь к базе данных ‘items’ через переменную ‘database’
• Получите доступ к коллекциям в базе данных с помощью переменной ‘item_collection’

Создание модели базы данных с помощью pydantic

В этом разделе вы будете проектировать свои данные с помощью ‘ pydantic ‘, который предоставляет моделирование для нашей базы данных MongoDB.

Установите модуль ‘pydantic’ с помощью команды ‘pip3’ ниже. Модуль ‘ pydantic ‘ — это библиотека валидации данных, которая позволяет вам создавать схемы базы данных через модель.

pip3 install pydantic

Теперь откройте файл ‘server/models/itemModels.py‘ с помощью текстового редактора и скопируйте следующий скрипт.

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,  
 }  
 }

• Импортируйте ‘BaseModel’ и ‘Field’ модули из ‘pydantic’
• Импортируйте модуль ‘Optional’ из ‘typing’
• Создайте следующую схему базы данных ‘Item’ для FastAPI: - ‘name’ и ‘category’ с типом строка
• ‘stocks’ и ‘price’ с типом целое число
• ‘price’ должен быть больше 0
• Расширьте модель данных через класс ‘Config’, предоставив пример данных, которые пользователь может включить в запрос
• Создайте следующую схему ‘ItemUpdate’ с каждым полем как необязательным.

Добавление операций CRUD

На данный момент вы создали подключение к серверу MongoDB и создали схему базы данных через ‘pydantic’. Давайте перейдем к созданию операций CRUD с FastAPI и MongoDB.

Снова откройте скрипт ‘ server/database.py ‘ и создайте новую функцию ‘ item_helper ‘ с типом ‘ dict ‘. Вызовите эту функцию, чтобы получить подробные данные о предмете.

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

Добавьте следующую строку, чтобы импортировать модуль ‘ ObjectId ‘ из ‘bson.objectid’. ‘ ObjectId ‘ является частью типа данных в BSON, который MongoDB использует для хранения данных и представления данных JSON.

from bson.objectid import ObjectId

Создание элемента

Сначала вы создадите асинхронную функцию, которая сможет добавлять новые данные в базу данных MongoDB.

Создайте новую функцию ‘a dd_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)

• Функция ‘add_item’ принимает словарные данные ваших деталей элемента
• Через ‘item_collection’ получить доступ к документам в базе данных и выполнить запрос ‘insert_one’, чтобы добавить новый элемент
• Новый элемент будет выведен при успешной операции

Получить все элементы

Во-вторых, вы создадите новую функцию ‘get_items’, которая сможет извлекать все доступные элементы в вашей базе данных.

Создайте новую функцию ‘ get_items ‘ для извлечения всех элементов из MongoDB.

# извлечь все элементы  
async def get_items():  
 items = []  
 async for item in item_collection.find():  
 items.append(item_helper(item))  
 return items

• Вы создадите пустой список ‘items’
• Используя запрос ‘find()’, чтобы искать все данные и проходить по ним
• Каждый элемент будет добавлен в список ‘items’ с помощью метода ‘append’
• После завершения цикла ваши элементы будут показаны

Получить конкретный элемент по id

Далее вы создадите новую функцию, которая сможет извлекать данные для конкретного селектора ‘id’. Это покажет вам подробные данные о конкретных элементах.

Создайте новую функцию ‘ get_item ‘, чтобы извлечь данные конкретных элементов. Функция ‘get_item’ будет принимать строку ‘id’ в качестве селектора и возвращать словарь.

# извлечь конкретный элемент  
async def get_item(id: str) -> dict:  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 return item_helper(item)  
 return "Элемент не найден."

• Запрос ‘find_one()’ будет использоваться для извлечения данных на основе ‘id’
• Если элемент найден, детали будут показаны с использованием формата словаря ‘item_helper’
• Если элемент недоступен, ответ будет ‘Элемент не найден’

Обновить элемент

Теперь вы создадите новую функцию для обновления элемента. Вы также можете частично обновить ваш элемент через API.

Определите новую функцию ‘ change_item ‘ следующим образом:

# обновить элемент  
async def change_item(id: str, data: dict):  
 if len(data) < 1:  
 return "Пожалуйста, введите ваши данные"  
 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

• Функция ‘change_item’ принимает два аргумента, ‘id’ целевого элемента и ‘data’ как новые данные
• Если данные пустые или ‘< 1’, операция прекращается
• Эта функция найдет элемент на основе селектора ‘id’
• Если ‘id’ найден, будет выполнен ‘item_update’
• Если ‘item_update’ успешен, вернется ‘True’, или вернется ‘False’, если нет

Удалить элемент

Наконец, вы создадите функцию ‘delete_item’ для удаления элементов через конкретный селектор.

Чтобы удалить элемент, мы создадим функцию ‘ delete_item ‘ и используем селектор ‘id’ следующим образом:

# удалить элемент  
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'Элемент {id} удален.')  
 return "Элемент не найден."

• Запрос ‘find_one()’ будет искать элемент на основе предоставленного ‘id’
• Если элемент найден, он будет удален через запрос ‘delete_one()’ и вернет ‘Элемент id удален’
• Если элемент недоступен, будет показано ‘Элемент не найден’

Добавление маршрутов для операций CRUD

На данный момент мы создали асинхронные функции для операций CRUD с FastAPI. Теперь давайте создадим маршрут или конечную точку для каждой операции.

Отредактируйте файл ‘server/routes/item.py‘ с помощью вашего предпочтительного редактора.

server/routes/item.py

Сначала добавьте модули ‘APIRouter‘ и ‘Body‘ из ‘fastapi’. Затем добавьте ‘jsonable_encode‘ из ‘fastapi.encoders’.

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

Импортируйте каждую функцию из файла ‘database.py’.

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

Импортируйте модели ‘Item’ и ‘ItemUpdate’ из файла ‘itemModels.py’.

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

Вызовите класс ‘APIRouter’ через переменную ‘router’.

router = APIRouter()

Маршрут для добавления нового элемента

Теперь давайте добавим маршрут для добавления новых элементов. В этом примере вы можете добавить новый элемент через POST на URL ‘/item‘.

Добавьте следующие строки, чтобы настроить маршруты для добавления новых элементов. Каждый POST на корневой URL элемента будет рассматриваться как вставка новых данных.

# операция добавления нового элемента  
@router.post("/")  
async def add_item_data(item: Item = Body(...)):  
 item = jsonable_encoder(item)  
 new_item = await add_item(item)  
 return new_item

• Функция ‘add_item_data’ принимает схему ‘Item’, которая будет частью ‘Body’ в вашем запросе
• Ваш элемент будет преобразован в словарный формат через ‘jsonable_encoder’
• Вставьте ваши словарные данные через функцию ‘add_item’ (см. database.py) в переменную ‘new_item’
• Верните ваши вставленные данные ‘new_item’ в качестве ответа

Маршрут для получения всех элементов

Добавьте следующий скрипт, чтобы настроить маршрут для извлечения данных. Каждый GET-запрос к корневому URL элемента извлечет все ваши данные.

# получить все доступные элементы  
@router.get("/")  
async def get_item_data():  
 items = await get_items()  
 if items:  
 return items  
 return "Нет доступных элементов."

• Создайте функцию ‘get_item_data’, которая выполнит функцию ‘get_item’ из файла ‘database.py’
• Если элементы доступны, вы получите список всех ваших элементов
• Если нет элемента, вы получите ответ ‘Нет доступных элементов’

Маршрут для получения конкретного элемента

Чтобы получить детали конкретного элемента, мы будем использовать ‘id’ в качестве селектора. Каждый GET-запрос к ‘/id’ вернет детализированный элемент запрашиваемого ‘id’.

# Показать детали элемента по id  
@router.get("/{id}")  
async def get_item_details(id):  
 item_details = await get_item(id)  
 if item_details:  
 return item_details  
 return "Элемент не найден."

• Функция ‘get_item_details’ будет создана и передаст ‘id’ из URL
• Функция ‘get_item’ (см. database.py) будет вызвана и также передаст ‘id’ в качестве аргумента
• Если элемент найден, будут показаны детали элемента
• Если нет элемента с этим конкретным ‘id’, вы получите ‘Элемент не найден’

Маршрут для обновления элемента

Скопируйте следующий код, чтобы установить маршрут для обновленного элемента:

# Обновить элемент  
@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'Успех: элемент {id} обновлен.'}  
 return "Ошибка"

• Функция ‘update_item’ будет принимать два аргумента, ‘id’ в качестве селектора и ‘data’, которая основана на модели ‘ItemUpdate’
• Данные будут проверены в переменной ‘data’
• ‘updated_item’ выполнит функцию ‘change_item’ из файла ‘database.py’
• Если обновление успешно, вы увидите вывод ‘Успех’

Маршрут для удаления элемента

Вставьте следующие строки, чтобы создать функцию ‘remove_item’ для удаления элементов. Каждая операция DELETE на конкретном ‘/id’ удалит элемент, соответствующий ‘id’.

# удалить элемент по 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'Элемент {id} недоступен.'}

• Функция ‘remove_item’ выполнит ‘delete_item’ и передаст селектор ‘id’
• Операция удаления будет сохранена и выполнена через переменную ‘item_to_delete’
• Когда элемент недоступен, вы получите ответ ‘Элемент id недоступен’

server/app.py

Теперь, когда вы завершили файл ‘server/routes/item.py’, давайте включим его в ‘server/app.py’.

Откройте файл ‘app.py‘ с помощью текстового редактора.

Импортируйте ‘router’ из файла ‘server/routes/item.py‘ как ‘ ItemRouter ‘.

from server.routes.item import router as ItemRouter

Вставьте ‘ItemRouter’ с префиксом по умолчанию ‘ /item ‘. Операции CRUD будут обрабатываться через URL ‘ /item ‘.

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

Теперь ваша конечная точка CRUD будет доступна по следующим адресам:

• Добавить новый элемент: POST на ‘ /item ‘
• Извлечь все элементы: GET на ‘ /item ‘
• Извлечь конкретный элемент: GET на ‘/item/id’. ‘id’ генерируется MongoDB
• Обновить элемент: PUT на ‘/item/id’
• Удалить элемент: DELETE на ‘/item/id’

Тестирование операций CRUD

Сначала убедитесь, что ваш проект FastAPI работает, или вы можете выполнить скрипт ‘main.py’ следующим образом:

python3 main.py

Перейдите по адресу http://SERVERIP:8080/docs, и вы увидите каждый маршрут, который вы создали.

Ниже приведен пример добавления нового элемента.

Извлеките все доступные элементы через API.

Извлеките конкретные элементы через селектор ‘id’.

Обновите частичные данные конкретных элементов.

Ниже приведены обновленные данные.

Ниже приведена операция удаления через селектор ‘id’.

Заключение

Поздравляем! Вы завершили установку FastAPI с MongoDB на Ubuntu 24.04. Вы также узнали, как подключить FastAPI к MongoDB с помощью модуля ‘motor’, создать модели данных с помощью ‘pydantic’, создать операции CRUD с FastAPI и научились создавать конечные точки для вашего API.