gongzhiyongandClaude Sonnet 4.6
433d2361a8
Litestar's {ticket_id:uuid} path parameter requires the handler
parameter to be typed as UUID, not str. This caused a 400 validation
error on GET /api/tickets/{uuid}.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
"""Ticket API endpoints — proxy to Gongdan system.
|
|
|
|
Returns data in a format aligned with the frontend TicketData interface:
|
|
{ id, title, status, priority, createdAt }
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from uuid import UUID
|
|
|
|
import httpx
|
|
from litestar import get
|
|
from litestar.exceptions import NotFoundException
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def _gongdan_headers() -> dict[str, str]:
|
|
return {"X-Api-Key": settings.gongdan_api_key}
|
|
|
|
|
|
def _map_status(raw: str) -> str:
|
|
mapping = {
|
|
"OPEN": "pending",
|
|
"ASSIGNED": "processing",
|
|
"IN_PROGRESS": "processing",
|
|
"PENDING_CUSTOMER": "processing",
|
|
"RESOLVED": "resolved",
|
|
"CLOSED": "resolved",
|
|
}
|
|
return mapping.get(raw, "pending")
|
|
|
|
|
|
def _map_priority(raw: str) -> str:
|
|
mapping = {
|
|
"URGENT": "P0",
|
|
"PRIORITY": "P1",
|
|
"NORMAL": "P2",
|
|
"LOW": "P3",
|
|
}
|
|
return mapping.get(raw, "P2")
|
|
|
|
|
|
def _transform_ticket(t: dict) -> dict:
|
|
"""Transform a Gongdan ticket to the frontend TicketData shape."""
|
|
return {
|
|
"id": t.get("id", ""),
|
|
"ticketNumber": t.get("ticketNumber", ""),
|
|
"title": t.get("description", "")[:120] or "No description",
|
|
"status": _map_status(t.get("status", "")),
|
|
"priority": _map_priority(t.get("priority", "")),
|
|
"createdAt": t.get("createdAt", ""),
|
|
}
|
|
|
|
|
|
@get("/api/tickets/summary")
|
|
async def get_tickets_summary() -> dict:
|
|
"""GET /api/tickets/summary -- Aggregate ticket statistics.
|
|
|
|
Returns:
|
|
{
|
|
"total": int,
|
|
"by_status": {"pending": N, "processing": N, "resolved": N},
|
|
"by_priority": {"P0": N, "P1": N, "P2": N, "P3": N}
|
|
}
|
|
"""
|
|
# Fetch a large page to get representative counts.
|
|
# For a production system this would be a dedicated aggregation API;
|
|
# the Gongdan API only exposes list endpoints so we aggregate client-side.
|
|
url = f"{settings.gongdan_api_base}/api/tickets"
|
|
params = {"page": 1, "pageSize": 100}
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(url, params=params, headers=_gongdan_headers())
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
tickets = data.get("tickets", [])
|
|
|
|
status_counter: Counter[str] = Counter()
|
|
priority_counter: Counter[str] = Counter()
|
|
|
|
for t in tickets:
|
|
status_counter[_map_status(t.get("status", ""))] += 1
|
|
priority_counter[_map_priority(t.get("priority", ""))] += 1
|
|
|
|
return {
|
|
"total": len(tickets),
|
|
"by_status": {
|
|
"pending": status_counter.get("pending", 0),
|
|
"processing": status_counter.get("processing", 0),
|
|
"resolved": status_counter.get("resolved", 0),
|
|
},
|
|
"by_priority": {
|
|
"P0": priority_counter.get("P0", 0),
|
|
"P1": priority_counter.get("P1", 0),
|
|
"P2": priority_counter.get("P2", 0),
|
|
"P3": priority_counter.get("P3", 0),
|
|
},
|
|
}
|
|
|
|
|
|
@get("/api/tickets")
|
|
async def list_tickets(page: int = 1, page_size: int = 20) -> list[dict]:
|
|
"""GET /api/tickets — List tickets from Gongdan, formatted for frontend."""
|
|
url = f"{settings.gongdan_api_base}/api/tickets"
|
|
params = {"page": page, "pageSize": page_size}
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(url, params=params, headers=_gongdan_headers())
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
tickets = data.get("tickets", [])
|
|
return [_transform_ticket(t) for t in tickets]
|
|
|
|
|
|
@get("/api/tickets/{ticket_id:uuid}")
|
|
async def get_ticket(ticket_id: UUID) -> dict:
|
|
"""GET /api/tickets/:id — Get a single ticket detail."""
|
|
url = f"{settings.gongdan_api_base}/api/tickets/{ticket_id}"
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(url, headers=_gongdan_headers())
|
|
if resp.status_code == 404:
|
|
raise NotFoundException(detail=f"Ticket {ticket_id} not found")
|
|
resp.raise_for_status()
|
|
t = resp.json()
|
|
|
|
return _transform_ticket(t)
|