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
|
|
|
|
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 == "pending"))
|
|
t_progress = await db.scalar(select(func.count(Task.id)).where(Task.status == "in_progress"))
|
|
t_done = await db.scalar(select(func.count(Task.id)).where(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,
|
|
)
|