建立一个简单的todo应用步伐（前端React；后端FastAPI；数据库MongoDB） - ToB企服应用市场:ToB评测及商务社交产业平台

mkdir farm-todo
cd farm-todo

复制代码

mkdir frontend backend
cd backend

复制代码

python -m venv venv
source venv/bin/activate

复制代码

pip install "fastapi[all]" "motor[srv]" beanie aiostream

复制代码

pip freeze > requirements.txt

复制代码

FROM python:3
WORKDIR /usr/src/app
COPY requirements.txt ./
RUN pip install --no-cache-dir --upgrade -r ./requirements.txt
EXPOSE 3001
CMD [ "python", "./src/server.py" ]

复制代码

[tool.pytest.ini_options]
pythonpath = "src"

复制代码

from bson import ObjectId
# ObjectId是MongoDB的默认主键类型，通常用于在查询和处理数据库中的唯一标识符。
from motor.motor_asyncio import AsyncIOMotorCollection
# motor是MongoDB的异步驱动程序，用于与MongoDB进行异步交互。
# AsyncIOMotorCollection是一个集合（collection）对象，提供了对MongoDB集合的异步操作。
from pymongo import ReturnDocument
# ReturnDocument 是pymongo中的常量，用于在更新操作时指定返回更新前或更新后的文档。
from pydantic import BaseModel
# Pydantic的BaseModel类是一个数据验证和数据结构定义工具。
from uuid import uuid4
class ListSummary(BaseModel):
id: str # 文档的唯一标识符，将MongoDB中的ObjectId转换为字符串。
name: str # 文档的名称字段。
item_count: int # 文档中项目的数量字段。
@staticmethod
def from_doc(doc) -> "ListSummary":
return ListSummary(
id=str(doc["_id"]),
name=doc["name"],
item_count=doc["item_count"],
)
# 这是一个静态方法，接受一个MongoDB文档（字典类型）作为参数，
# 并将该文档转换为ListSummary对象。doc通常是从MongoDB集合中查询返回的文档数据。
class ToDoListItem(BaseModel):
id: str
label: str
checked: bool
@staticmethod
def from_doc(item) -> "ToDoListItem":
return ToDoListItem(
id=item["id"],
label=item["label"],
checked=item["checked"],
)
# 将MongoDB中表示单个待办事项的文档（item）转换为ToDoListItem对象。
# 它提取id、label和checked字段，并使用这些字段来创建一个ToDoListItem实例。
class ToDoList(BaseModel):
id: str
name: str
items: list[ToDoListItem]
@staticmethod
def from_doc(doc) -> "ToDoList":
return ToDoList(
id=str(doc["_id"]),
name=doc["name"],
items=[ToDoListItem.from_doc(item) for item in doc["items"]],
)
# 接收一个待办事项列表的MongoDB文档（doc），将其转换为ToDoList对象。
# 方法从doc中提取_id和name字段，并将items字段中的每个待办事项文档
# 通过ToDoListItem.from_doc方法转换为ToDoListItem对象，最终返回一个ToDoList实例。
# 处理对MongoDB中待办事项数据的增删改查操作。
class ToDoDAL:
def __init__(self, todo_collection: AsyncIOMotorCollection):
self._todo_collection = todo_collection
# 返回所有待办事项列表的简要信息（名称和项目数）
async def list_todo_lists(self, session=None):
async for doc in self._todo_collection.find(
{},
projection={
"name": 1,
"item_count": {"$size": "$items"},
},
sort={"name": 1},
session=session,
):
yield ListSummary.from_doc(doc)
# 创建一个新的待办事项列表。
# 插入文档的_id字符串形式。
async def create_todo_list(self, name: str, session=None) -> str:
response = await self._todo_collection.insert_one(
{"name": name, "items": []},
session=session,
)
return str(response.inserted_id)
# 根据ID获取特定待办事项列表的完整信息。
async def get_todo_list(self, id: str | ObjectId, session=None) -> ToDoList:
doc = await self._todo_collection.find_one(
{"_id": ObjectId(id)},
session=session,
)
return ToDoList.from_doc(doc)
# 根据ID删除一个待办事项列表。
async def delete_todo_list(self, id: str | ObjectId, session=None) -> bool:
response = await self._todo_collection.delete_one(
{"_id": ObjectId(id)},
session=session,
)
return response.deleted_count == 1
# 向指定待办事项列表中添加新项目
async def create_item(
self,
id: str | ObjectId,
label: str,
session=None,
) -> ToDoList | None:
result = await self._todo_collection.find_one_and_update(
{"_id": ObjectId(id)},
{
"$push": {
"items": {
"id": uuid4().hex,
"label": label,
"checked": False,
}
}
},
session=session,
return_document=ReturnDocument.AFTER,
)
if result:
return ToDoList.from_doc(result)
# 设置待办事项item的checked状态。
async def set_checked_state(
self,
doc_id: str | ObjectId,
item_id: str,
checked_state: bool,
session=None,
) -> ToDoList | None:
result = await self._todo_collection.find_one_and_update(
{"_id": ObjectId(doc_id), "items.id": item_id},
{"$set": {"items.$.checked": checked_state}},
session=session,
return_document=ReturnDocument.AFTER,
)
if result:
return ToDoList.from_doc(result)
# 从指定待办事项列表中删除一个项目。
async def delete_item(
self,
doc_id: str | ObjectId,
item_id: str,
session=None,
) -> ToDoList | None:
result = await self._todo_collection.find_one_and_update(
{"_id": ObjectId(doc_id)},
{"$pull": {"items": {"id": item_id}}},
session=session,
return_document=ReturnDocument.AFTER,
)
if result:
return ToDoList.from_doc(result)

复制代码

from contextlib import asynccontextmanager
from datetime import datetime
import os
import sys
from bson import ObjectId
from fastapi import FastAPI, status
from motor.motor_asyncio import AsyncIOMotorClient
# 是MongoDB异步客户端（motor库）来连接数据库。
from pydantic import BaseModel
import uvicorn # 是ASGI服务器，用于运行FastAPI应用。
from dal import ToDoDAL, ListSummary, ToDoList
COLLECTION_NAME = "todo_lists"
MONGODB_URI = os.environ["MONGODB_URI"]
# 从环境变量中获取MongoDB的URI连接字符串。
DEBUG = os.environ.get("DEBUG", "").strip().lower() in {"1", "true", "on", "yes"}
# 从环境变量读取调试模式，接受多种True的写法。
# 这是一个异步上下文管理器，管理FastAPI应用的启动和关闭。
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup:
client = AsyncIOMotorClient(MONGODB_URI)
# 创建MongoDB异步客户端实例，连接到数据库。
database = client.get_default_database()
# 获取MongoDB数据库实例，默认使用连接字符串中的数据库名称。
# ping命令用于测试MongoDB连接是否正常。pong["ok"] == 1表示连接成功。
pong = await database.command("ping")
if int(pong["ok"]) != 1:
raise Exception("Cluster connection is not okay!")
todo_lists = database.get_collection(COLLECTION_NAME)
# 获取待办事项集合todo_lists。
app.todo_dal = ToDoDAL(todo_lists)
# 将ToDoDAL实例附加到FastAPI应用实例上，便于其他模块使用。
# Yield back to FastAPI Application:
yield
# 暂停上下文管理器，将控制权交给应用，以继续运行其他操作。
# Shutdown:
client.close()
# 应用关闭时，自动关闭MongoDB连接以释放资源。
app = FastAPI(lifespan=lifespan, debug=DEBUG)
# 使用lifespan管理器（在应用启动和关闭时管理数据库连接）。
# debug=DEBUG设定调试模式，使得在调试时更容易看到错误和详细信息。
# 异步获取所有待办事项列表的摘要信息。
@app.get("/api/lists")
async def get_all_lists() -> list[ListSummary]:
return [i async for i in app.todo_dal.list_todo_lists()]
# 用于创建新的待办事项列表，包含列表名称。
class NewList(BaseModel):
name: str
# 用于创建操作成功后的响应，包含列表的id和name。
class NewListResponse(BaseModel):
id: str
name: str
# 接受NewList格式的数据，创建新的待办事项列表。
@app.post("/api/lists", status_code=status.HTTP_201_CREATED)
async def create_todo_list(new_list: NewList) -> NewListResponse:
return NewListResponse(
id=await app.todo_dal.create_todo_list(new_list.name),
name=new_list.name,
)
# 使用列表的唯一标识符list_id获取单个待办事项列表。
# 调用get_todo_list方法，返回一个包含列表详细信息的ToDoList对象。
@app.get("/api/lists/{list_id}")
async def get_list(list_id: str) -> ToDoList:
"""Get a single to-do list"""
return await app.todo_dal.get_todo_list(list_id)
# 根据list_id删除指定的待办事项列表。
# 调用delete_todo_list方法，返回布尔值表示删除是否成功。
@app.delete("/api/lists/{list_id}")
async def delete_list(list_id: str) -> bool:
return await app.todo_dal.delete_todo_list(list_id)
# 用于添加新待办事项的数据模型，包含label字段，表示待办事项的标签（描述）。
class NewItem(BaseModel):
label: str
# 用于返回新添加项的响应模型，包含id和label字段。
class NewItemResponse(BaseModel):
id: str
label: str
# 用于向指定的待办事项列表（list_id）中添加新待办事项items。
@app.post(
"/api/lists/{list_id}/items/",
status_code=status.HTTP_201_CREATED,
)
# 调用create_item方法，将list_id和new_item.label传入。
# 返回更新后的完整待办事项列表ToDoList，包含新的待办事项item。
async def create_item(list_id: str, new_item: NewItem) -> ToDoList:
return await app.todo_dal.create_item(list_id, new_item.label)
# 根据list_id和item_id删除指定待办事项列表中的特定项。
# 返回更新后的ToDoList对象，删除项后更新的列表。
@app.delete("/api/lists/{list_id}/items/{item_id}")
async def delete_item(list_id: str, item_id: str) -> ToDoList:
return await app.todo_dal.delete_item(list_id, item_id)
# 定义更新待办事项item的完成状态所需的数据模型。
# 包含item_id（待办事项item的唯一标识符）和checked_state（布尔值，表示是否已完成）字段。
class ToDoItemUpdate(BaseModel):
item_id: str
checked_state: bool
# 用于更新指定待办事项列表中某个待办事项item的完成状态。
# 返回更新后的ToDoList对象，反映状态变更后的完整列表。
@app.patch("/api/lists/{list_id}/checked_state")
async def set_checked_state(list_id: str, update: ToDoItemUpdate) -> ToDoList:
return await app.todo_dal.set_checked_state(
list_id, update.item_id, update.checked_state
)
class DummyResponse(BaseModel):
id: str
when: datetime
@app.get("/api/dummy")
async def get_dummy() -> DummyResponse:
return DummyResponse(
id=str(ObjectId()),
when=datetime.now(),
)
# 使用ObjectId()生成一个新的唯一标识符并转换为字符串，赋值给id。
# 每个文档在MongoDB数据库中都有一个_id字段，默认情况下，它的值是一个ObjectId。
# 使用datetime.now()获取当前时间并赋值给when。
def main(argv=sys.argv[1:]):
try:
# 使用uvicorn来启动FastAPI应用
uvicorn.run("server:app", host="0.0.0.0", port=3001, reload=DEBUG)
# host="0.0.0.0"：让应用在所有可用的网络接口上监听。
# reload=DEBUG：如果DEBUG为True，在代码更改时自动重载应用（适合开发阶段）。
except KeyboardInterrupt:
pass
# 使用try...except来捕获KeyboardInterrupt异常，以便在按下Ctrl+C时优雅地停止服务。
if __name__ == "__main__":
main()