DevGang
Авторизоваться

Python 3 + FastAPI + MongoDB

В этой статье мы рассмотрим Python и FastAPI, объединив персистентность с MongoDB. Также мы посмотрим, как использовать MongoDB с Docker.

Первый шаг в настройке MongoDB с помощью Docker включает проверку работоспособности Docker в вашей системе. Убедившись, что Docker активен, вы можете двигаться дальше, создав контейнер базы данных MongoDB. Для этого выполните следующую команду в своем терминале:

docker run --name mongo-local -d -p 27017:27017 mongo

Теперь для работы с MongoDB нам нужно установить пакет. В этой статье мы будем использовать двигатель.

pip3 install motor

С этими шагами все в порядке, давайте создадим новый файл Python с именем db.py для создания нашего репозитория:

from motor.motor_asyncio import AsyncIOMotorClient
from bson import ObjectId


class DBManager:
    def __init__(self, uri: str, database_name: str):
        self.client = AsyncIOMotorClient(uri)
        self.db = self.client[database_name]

    async def create_item(self, item):
        result = await self.db.items.insert_one(item)
        return await self.read_item(result.inserted_id)

    async def read_item(self, item_id: ObjectId):
        item = await self.db.items.find_one({"_id": item_id})
        if item:
            item["id"] = str(item["_id"])
            del item["_id"]
        return item

    async def read_items(self) -> list:
        items = []
        cursor = self.db.items.find({})
        async for item in cursor:
            item["id"] = str(item["_id"])
            del item["_id"]
            items.append(item)
        return items

    async def update_item(self, item_id: ObjectId, item):
        await self.db.items.update_one({"_id": item_id}, {"$set": item})
        return await self.read_item(item_id)

    async def delete_item(self, item_id: ObjectId) -> bool:
        result = await self.db.items.delete_one({"_id": item_id})
        return result.deleted_count > 0


db_manager = DBManager(uri="mongodb://localhost:27017", database_name="db_local")

В этом коде мы создаем класс DBManager, в котором мы разработали пять методов: create_item, read_item, read_items, update_item и delete_item.

Теперь мы можем обновить наш файл main.py с помощью этого кода:

from bson import ObjectId
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from uuid import UUID, uuid4
from db import db_manager
from typing import List


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    on_offer: bool = False


app = FastAPI()


@app.post("/items/", response_model=Item)
async def create_item(item: Item):   
    new_item = await db_manager.create_item(item.model_dump())
    return new_item


@app.get("/items/", response_model=List[Item])
async def read_items():
    return await db_manager.read_items()


@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    item = await db_manager.read_item(ObjectId(item_id))
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item


@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    updated_item = await db_manager.update_item(ObjectId(item_id), item.model_dump())
    if updated_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return updated_item


@app.delete("/items/{item_id}", response_model=str)
async def delete_item(item_id: str):
    deleted = await db_manager.delete_item(ObjectId(item_id))
    if not deleted:
        raise HTTPException(status_code=404, detail="Item not found")
    return "Item deleted successfully"       

В этом коде мы просто добавляем db_manager для хранения всех данных в mongodb, а не в памяти.

Чтобы протестировать свой проект, выполните в терминале следующую команду.

uvicorn main:app --reload.

Это позволит вам протестировать все действия в Swagger и просмотреть данные, хранящиеся в вашей базе данных MongoDB.

Источник:

#Python #MongoDB #FastAPI #Docker
Комментарии
Чтобы оставить комментарий, необходимо авторизоваться

Присоединяйся в тусовку

В этом месте могла бы быть ваша реклама

Разместить рекламу