PTah
100 lines
2.8 KiB
Python
100 lines
2.8 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.auth.jwt_auth import get_current_user
|
|
from app.database import get_db
|
|
from app.models import Problem
|
|
|
|
router = APIRouter(prefix="/problems", tags=["problems"])
|
|
|
|
|
|
class ProblemSummary(BaseModel):
|
|
id: int
|
|
host_id: int | None
|
|
hostname: str | None
|
|
title: str
|
|
summary: str
|
|
severity: str
|
|
status: str
|
|
rule_id: str | None
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class ProblemListResponse(BaseModel):
|
|
items: list[ProblemSummary]
|
|
total: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
@router.get("", response_model=ProblemListResponse)
|
|
def list_problems(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
status: str | None = Query(None),
|
|
db: Session = Depends(get_db),
|
|
_user: str = Depends(get_current_user),
|
|
) -> ProblemListResponse:
|
|
stmt = select(Problem).options(joinedload(Problem.host))
|
|
count_stmt = select(func.count()).select_from(Problem)
|
|
if status:
|
|
stmt = stmt.where(Problem.status == status)
|
|
count_stmt = count_stmt.where(Problem.status == status)
|
|
|
|
total = db.scalar(count_stmt) or 0
|
|
rows = db.scalars(
|
|
stmt.order_by(Problem.updated_at.desc()).offset((page - 1) * page_size).limit(page_size)
|
|
).all()
|
|
|
|
items = [
|
|
ProblemSummary(
|
|
id=p.id,
|
|
host_id=p.host_id,
|
|
hostname=p.host.hostname if p.host else None,
|
|
title=p.title,
|
|
summary=p.summary,
|
|
severity=p.severity,
|
|
status=p.status,
|
|
rule_id=p.rule_id,
|
|
created_at=p.created_at,
|
|
updated_at=p.updated_at,
|
|
)
|
|
for p in rows
|
|
]
|
|
return ProblemListResponse(items=items, total=total, page=page, page_size=page_size)
|
|
|
|
|
|
@router.post("/{problem_id}/ack")
|
|
def ack_problem(
|
|
problem_id: int,
|
|
db: Session = Depends(get_db),
|
|
_user: str = Depends(get_current_user),
|
|
) -> dict:
|
|
problem = db.get(Problem, problem_id)
|
|
if problem is None:
|
|
raise HTTPException(status_code=404, detail="Problem not found")
|
|
problem.status = "acknowledged"
|
|
db.commit()
|
|
return {"id": problem.id, "status": problem.status}
|
|
|
|
|
|
@router.post("/{problem_id}/resolve")
|
|
def resolve_problem(
|
|
problem_id: int,
|
|
db: Session = Depends(get_db),
|
|
_user: str = Depends(get_current_user),
|
|
) -> dict:
|
|
problem = db.get(Problem, problem_id)
|
|
if problem is None:
|
|
raise HTTPException(status_code=404, detail="Problem not found")
|
|
problem.status = "resolved"
|
|
db.commit()
|
|
return {"id": problem.id, "status": problem.status}
|