duxingchen
69f3e35d14
1. Dashboard统计Bug修复 - Task统计改用TASK_STATUS_PENDING/WIP/COMPLETED大写常量 - 旧代码使用小写"pending"/"in_progress"永远匹配不到数据 - 修复后tasks_pending/tasks_in_progress/tasks_completed返回真实值 2. 生产环境SECRET_KEY强制校验 - 新增model_validator：DEBUG=False且SECRET_KEY为默认值时抛出ValueError - 阻止使用默认密钥部署到生产环境
42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
"""Dashboard 统计服务"""
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class DashboardStats(BaseModel):
|
|
products_total: int
|
|
products_pending: int
|
|
products_in_progress: int
|
|
products_completed: int
|
|
tasks_total: int
|
|
tasks_pending: int
|
|
tasks_in_progress: int
|
|
tasks_completed: int
|
|
|
|
|
|
async def get_dashboard_stats(db: AsyncSession) -> DashboardStats:
|
|
from app.models.product import Product
|
|
from app.models.task import Task, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED
|
|
|
|
p_total = await db.scalar(select(func.count(Product.id)))
|
|
p_pending = await db.scalar(select(func.count(Product.id)).where(Product.status == "pending"))
|
|
p_progress = await db.scalar(select(func.count(Product.id)).where(Product.status == "in_progress"))
|
|
p_done = await db.scalar(select(func.count(Product.id)).where(Product.status == "completed"))
|
|
|
|
t_total = await db.scalar(select(func.count(Task.id)))
|
|
t_pending = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_PENDING))
|
|
t_progress = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_WIP))
|
|
t_done = await db.scalar(select(func.count(Task.id)).where(Task.status == TASK_STATUS_COMPLETED))
|
|
|
|
return DashboardStats(
|
|
products_total=p_total or 0,
|
|
products_pending=p_pending or 0,
|
|
products_in_progress=p_progress or 0,
|
|
products_completed=p_done or 0,
|
|
tasks_total=t_total or 0,
|
|
tasks_pending=t_pending or 0,
|
|
tasks_in_progress=t_progress or 0,
|
|
tasks_completed=t_done or 0,
|
|
)
|