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.
