Spaces:
Sleeping
Sleeping
File size: 2,560 Bytes
c496840 f927995 c496840 f927995 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """
Endpoints for pharmaceutical supply chain alerts and monitoring.
"""
from datetime import datetime
from typing import List, Optional, Annotated, Any
from fastapi import APIRouter, Depends, Query, HTTPException, Response
from ..deps import get_data_loader
from ..models.alert import Alert, AlertSeverity, AlertSource, AlertListResponse
router = APIRouter(prefix="/api/v1", tags=["alerts"])
@router.get("/alerts", response_model=AlertListResponse)
async def get_alerts(
response: Response,
severity: Optional[AlertSeverity] = None,
drug_id: Optional[str] = None,
source: Optional[AlertSource] = None,
since: Optional[datetime] = None,
limit: int = Query(default=50, le=100),
offset: int = Query(default=0, ge=0),
data_loader: Annotated[Any, Depends(get_data_loader)] = None
) -> AlertListResponse:
"""
Returns a paginated list of supply chain alerts with multi-dimensional filtering.
Caching: The response includes a 10-second Cache-Control header to facilitate polling.
"""
# 1. Fetch filtered alerts from DataLoader (handles severity and drug_id)
# We use a higher limit to pull enough for secondary local filtering
raw_alerts = data_loader.get_alerts(
severity=severity,
drug_id=drug_id,
limit=1000
)
# 2. Apply secondary filters (source, since) and sort
filtered_alerts = []
for alert in raw_alerts:
if source and alert.source != source:
continue
if since and alert.created_at < since:
continue
filtered_alerts.append(alert)
# Sort by created_at descending (newest first)
sorted_alerts = sorted(filtered_alerts, key=lambda x: x.created_at, reverse=True)
# 3. Paginate
total = len(sorted_alerts)
paginated = sorted_alerts[offset : offset + limit]
# 4. Set Cache-Control header
response.headers["Cache-Control"] = "max-age=10"
return AlertListResponse(
alerts=paginated,
total=total,
page=(offset // limit) + 1,
page_size=limit
)
@router.get("/alert/{alert_id}", response_model=Alert)
async def get_alert_detail(
alert_id: str,
data_loader: Annotated[Any, Depends(get_data_loader)] = None
) -> Alert:
"""
Returns full details for a specific alert by its ID.
"""
alert = data_loader.get_alert(alert_id)
if not alert:
raise HTTPException(
status_code=404,
detail=f"Alert with ID '{alert_id}' not found"
)
return alert
|