62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
from fastapi import APIRouter, HTTPException, Query, BackgroundTasks
|
|
from models.schemas import EventData, EventResponse, EventsResponse, Event
|
|
from services.redis_client import redis_client
|
|
from datetime import datetime
|
|
import json
|
|
|
|
router = APIRouter()
|
|
|
|
@router.post("/publish-event",
|
|
response_model=EventResponse,
|
|
summary="Publish Event",
|
|
description="Publish an event to the Redis message bus",
|
|
responses={
|
|
200: {"description": "Event published successfully"},
|
|
500: {"description": "Failed to publish event"}
|
|
},
|
|
tags=["Events"])
|
|
async def publish_event(event_data: EventData, background_tasks: BackgroundTasks):
|
|
"""Publish an event to the Redis message bus for consumption by other services"""
|
|
try:
|
|
event = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"service": event_data.service,
|
|
"event_type": event_data.event_type,
|
|
"metadata": json.dumps(event_data.metadata)
|
|
}
|
|
|
|
# Publish to Redis
|
|
redis_client.lpush("events", json.dumps(event))
|
|
|
|
return EventResponse(
|
|
status="published",
|
|
event=event
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.get("/events",
|
|
response_model=EventsResponse,
|
|
summary="Get Events",
|
|
description="Retrieve recent events from the message bus",
|
|
responses={
|
|
200: {"description": "Successfully retrieved events"},
|
|
500: {"description": "Failed to retrieve events"}
|
|
},
|
|
tags=["Events"])
|
|
async def get_events(limit: int = Query(100, ge=1, le=1000, description="Maximum number of events to retrieve")):
|
|
"""Get recent events from the Redis message bus"""
|
|
try:
|
|
events = redis_client.lrange("events", 0, limit - 1)
|
|
parsed_events = []
|
|
for event in events:
|
|
try:
|
|
event_data = json.loads(event)
|
|
parsed_events.append(Event(**event_data))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
return EventsResponse(events=parsed_events)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|