25 lines
1.5 KiB
Python
25 lines
1.5 KiB
Python
|
|
|
|
# ==================== 统计 API ====================
|
|
|
|
@app.get("/stats/overview")
|
|
async def get_stats_overview(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
total_agents = db.query(sqlfunc.count(Agent.id)).scalar()
|
|
running_agents = db.query(sqlfunc.count(Agent.id)).filter(Agent.status == AgentStatus.running).scalar()
|
|
stopped_agents = db.query(sqlfunc.count(Agent.id)).filter(Agent.status == AgentStatus.stopped).scalar()
|
|
total_templates = db.query(sqlfunc.count(Agent.template_name.distinct())).scalar()
|
|
return {"total_agents": total_agents, "running_agents": running_agents, "stopped_agents": stopped_agents, "total_templates": total_templates}
|
|
|
|
@app.get("/stats/by-template")
|
|
async def get_stats_by_template(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
result = db.query(Agent.template_name, sqlfunc.count(Agent.id).label("agent_count"), sqlfunc.sum(Agent.current_replicas).label("total_replicas")).group_by(Agent.template_name).all()
|
|
return [{"template_name": r[0], "agent_count": r[1], "total_replicas": r[2] or 0} for r in result]
|
|
|
|
@app.get("/stats/by-owner")
|
|
async def get_stats_by_owner(db: Session = Depends(get_db)):
|
|
from sqlalchemy import func as sqlfunc
|
|
result = db.query(Agent.owner_id, sqlfunc.count(Agent.id).label("agent_count"), sqlfunc.sum(Agent.current_replicas).label("total_replicas")).group_by(Agent.owner_id).all()
|
|
return [{"owner_id": r[0], "agent_count": r[1], "total_replicas": r[2] or 0} for r in result]
|