102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
"""
|
||
业务逻辑层 —— 使用内存假数据实现 Todo 的 CRUD 操作。
|
||
|
||
所有数据存储在内存字典中，服务重启后数据会丢失。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
from pydantic import BaseModel, Field
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# Pydantic 模型（请求 / 响应）
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TodoCreate(BaseModel):
|
||
"""创建待办事项的请求体。"""
|
||
title: str = Field(..., min_length=1, max_length=200, description="待办事项标题")
|
||
completed: bool = Field(False, description="是否已完成")
|
||
|
||
|
||
class TodoUpdate(BaseModel):
|
||
"""更新待办事项的请求体（部分更新）。"""
|
||
title: Optional[str] = Field(None, min_length=1, max_length=200, description="待办事项标题")
|
||
completed: Optional[bool] = Field(None, description="是否已完成")
|
||
|
||
|
||
class TodoResponse(BaseModel):
|
||
"""待办事项的响应体。"""
|
||
id: int = Field(..., description="待办事项 ID")
|
||
title: str = Field(..., description="待办事项标题")
|
||
completed: bool = Field(..., description="是否已完成")
|
||
|
||
model_config = {"from_attributes": True}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 内存假数据存储
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class TodoService:
|
||
"""待办事项服务 —— 使用内存字典存储数据。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._todos: dict[int, dict] = {}
|
||
self._next_id: int = 1
|
||
self._init_mock_data()
|
||
|
||
def _init_mock_data(self) -> None:
|
||
"""初始化一批假数据，方便演示。"""
|
||
mock_todos = [
|
||
{"title": "学习 FastAPI", "completed": False},
|
||
{"title": "编写单元测试", "completed": False},
|
||
{"title": "阅读 Python 文档", "completed": True},
|
||
{"title": "完成项目报告", "completed": False},
|
||
{"title": "锻炼身体 30 分钟", "completed": True},
|
||
]
|
||
for item in mock_todos:
|
||
self.create_todo(TodoCreate(**item))
|
||
|
||
def list_todos(self) -> list[TodoResponse]:
|
||
"""返回所有待办事项。"""
|
||
return [TodoResponse(**todo) for todo in self._todos.values()]
|
||
|
||
def get_todo(self, todo_id: int) -> Optional[TodoResponse]:
|
||
"""根据 ID 获取单个待办事项，不存在则返回 None。"""
|
||
todo = self._todos.get(todo_id)
|
||
if todo is None:
|
||
return None
|
||
return TodoResponse(**todo)
|
||
|
||
def create_todo(self, todo_in: TodoCreate) -> TodoResponse:
|
||
"""创建新的待办事项并返回。"""
|
||
todo_id = self._next_id
|
||
self._next_id += 1
|
||
todo = {
|
||
"id": todo_id,
|
||
"title": todo_in.title,
|
||
"completed": todo_in.completed,
|
||
}
|
||
self._todos[todo_id] = todo
|
||
return TodoResponse(**todo)
|
||
|
||
def update_todo(self, todo_id: int, todo_in: TodoUpdate) -> Optional[TodoResponse]:
|
||
"""部分更新指定待办事项，不存在则返回 None。"""
|
||
todo = self._todos.get(todo_id)
|
||
if todo is None:
|
||
return None
|
||
update_data = todo_in.model_dump(exclude_unset=True)
|
||
todo.update(update_data)
|
||
return TodoResponse(**todo)
|
||
|
||
def delete_todo(self, todo_id: int) -> bool:
|
||
"""删除指定待办事项，成功返回 True，不存在返回 False。"""
|
||
if todo_id not in self._todos:
|
||
return False
|
||
del self._todos[todo_id]
|
||
return True
|