123 lines
3.2 KiB
Python
123 lines
3.2 KiB
Python
|
"""业务逻辑层模块。
|
|||
|
|
|
|||
|
|
提供任务管理的核心业务逻辑，使用内存存储模拟数据库操作。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
from app.models import TodoItem, TodoCreate, TodoUpdate
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TodoService:
|
|||
|
|
"""任务管理服务类。
|
|||
|
|
|
|||
|
|
使用内存字典存储任务数据，提供完整的增删改查操作。
|
|||
|
|
线程不安全，仅适用于单线程演示场景。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self) -> None:
|
|||
|
|
"""初始化服务，创建空的内存存储和 ID 计数器。"""
|
|||
|
|
self._store: dict[int, TodoItem] = {}
|
|||
|
|
self._next_id: int = 1
|
|||
|
|
|
|||
|
|
def _generate_id(self) -> int:
|
|||
|
|
"""生成下一个可用的任务 ID。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
自增的任务 ID。
|
|||
|
|
"""
|
|||
|
|
current = self._next_id
|
|||
|
|
self._next_id += 1
|
|||
|
|
return current
|
|||
|
|
|
|||
|
|
def _now(self) -> datetime:
|
|||
|
|
"""获取当前 UTC 时间。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
当前 UTC 时间戳。
|
|||
|
|
"""
|
|||
|
|
return datetime.now(timezone.utc)
|
|||
|
|
|
|||
|
|
def list_todos(self) -> tuple[list[TodoItem], int]:
|
|||
|
|
"""获取所有任务列表。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
(任务列表, 总数) 的元组。
|
|||
|
|
"""
|
|||
|
|
items = list(self._store.values())
|
|||
|
|
return items, len(items)
|
|||
|
|
|
|||
|
|
def get_todo(self, todo_id: int) -> Optional[TodoItem]:
|
|||
|
|
"""根据 ID 获取单个任务。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
todo_id: 任务 ID。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
如果找到返回 TodoItem，否则返回 None。
|
|||
|
|
"""
|
|||
|
|
return self._store.get(todo_id)
|
|||
|
|
|
|||
|
|
def create_todo(self, data: TodoCreate) -> TodoItem:
|
|||
|
|
"""创建新任务。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
data: 创建任务的请求数据。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
创建成功的 TodoItem。
|
|||
|
|
"""
|
|||
|
|
now = self._now()
|
|||
|
|
item = TodoItem(
|
|||
|
|
id=self._generate_id(),
|
|||
|
|
title=data.title,
|
|||
|
|
description=data.description,
|
|||
|
|
completed=False,
|
|||
|
|
created_at=now,
|
|||
|
|
updated_at=now,
|
|||
|
|
)
|
|||
|
|
self._store[item.id] = item
|
|||
|
|
return item
|
|||
|
|
|
|||
|
|
def update_todo(self, todo_id: int, data: TodoUpdate) -> Optional[TodoItem]:
|
|||
|
|
"""更新已有任务。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
todo_id: 要更新的任务 ID。
|
|||
|
|
data: 包含更新字段的请求数据。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
更新后的 TodoItem，如果任务不存在返回 None。
|
|||
|
|
"""
|
|||
|
|
existing = self._store.get(todo_id)
|
|||
|
|
if existing is None:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
update_dict = data.model_dump(exclude_none=True)
|
|||
|
|
if not update_dict:
|
|||
|
|
return existing
|
|||
|
|
|
|||
|
|
updated_item = existing.model_copy(update=update_dict)
|
|||
|
|
updated_item.updated_at = self._now()
|
|||
|
|
self._store[todo_id] = updated_item
|
|||
|
|
return updated_item
|
|||
|
|
|
|||
|
|
def delete_todo(self, todo_id: int) -> bool:
|
|||
|
|
"""删除指定任务。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
todo_id: 要删除的任务 ID。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
删除成功返回 True，任务不存在返回 False。
|
|||
|
|
"""
|
|||
|
|
if todo_id in self._store:
|
|||
|
|
del self._store[todo_id]
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 全局单例服务实例
|
|||
|
|
todo_service = TodoService()
|