# 导入所需模块
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from starlette.responses import JSONResponse
from motor.motor_asyncio import AsyncIOMotorClient
# 定义数据库URI
database_uri = "mongodb://localhost:27017"
# 定义FastAPI实例
app = FastAPI()
# 连接到MongoDB
client = AsyncIOMotorClient(database_uri)
database = client["fastapi-react"]
items = database.get_collection("items")
# 定义Item模型
class Item(BaseModel):
title: str
description: str = None
# 创建新项目
@app.post("/items/")
async def create_item(item: Item):
await items.insert_one(item.dict())
return JSONResponse(content={"message": "Item created successfully"}, status_code=201)
# 获取所有项目
@app.get("/items/")
async def get_items():
items_list = []
async for item in items.find():
items_list.append(item)
return items_list
# 获取单个项目
@app.get("/items/{item_id}")
async def get_item(item_id):
item = await items.find_one({"_id": item_id})
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
# 更新项目
@app.put("/items/{item_id}")
async def update_item(item_id, item: Item):
await items.update_one({"_id": item_id}, {"$set": item.dict()})
return JSONResponse(content={"message": "Item updated successfully"}, status_code=200)
# 删除项目
@app.delete("/items/{item_id}")
async def delete_item(item_id):
result = await items.delete_one({"_id": item_id})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="Item not found")
return JSONResponse(content={"message": "Item deleted successfully"}, status_code=200)在这个代码示例中，我们使用了FastAPI框架和Motor库与MongoDB进行异步交互。我们定义了一个Item模型来序列化和反序列化数据，并创建了用于创建、读取、更新和删除项目的路由。这个示例展示了如何在FastAPI应用中实现RESTful API，并使用异步编程模式提高性能。