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()
|