b7f18dcaec
- Add GET endpoint for individual arbitrage opportunities - Add POST endpoint for creating arbitrage opportunities - Add GET endpoint for individual trades - Add POST endpoint for creating trades - Add endpoints for system events (GET list, GET by ID, POST) - Update API router to include the new events endpoints - Fix linting issues
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
from typing import Any, Optional, List
|
|
from fastapi import APIRouter, Depends, Query, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
|
|
from app.db.session import get_db
|
|
from app.models.arbitrage import SystemEvent
|
|
from app.schemas.arbitrage import SystemEventCreate, SystemEvent as SystemEventSchema
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=List[SystemEventSchema])
|
|
async def get_system_events(
|
|
event_type: Optional[str] = Query(None, description="Filter by event type (startup, shutdown, error, warning, info)"),
|
|
component: Optional[str] = Query(None, description="Filter by component"),
|
|
limit: int = Query(50, ge=1, le=200, description="Number of events to return"),
|
|
offset: int = Query(0, ge=0, description="Pagination offset"),
|
|
db: Session = Depends(get_db)
|
|
) -> Any:
|
|
"""
|
|
Retrieve system events with optional filtering.
|
|
"""
|
|
query = db.query(SystemEvent)
|
|
|
|
# Apply filters
|
|
if event_type:
|
|
query = query.filter(SystemEvent.event_type == event_type)
|
|
|
|
if component:
|
|
query = query.filter(SystemEvent.component == component)
|
|
|
|
# Get paginated results
|
|
events = query.order_by(desc(SystemEvent.timestamp)).offset(offset).limit(limit).all()
|
|
|
|
return events
|
|
|
|
|
|
@router.get("/{event_id}", response_model=SystemEventSchema)
|
|
async def get_system_event(
|
|
event_id: int,
|
|
db: Session = Depends(get_db)
|
|
) -> Any:
|
|
"""
|
|
Get a specific system event by ID.
|
|
"""
|
|
event = db.query(SystemEvent).filter(SystemEvent.id == event_id).first()
|
|
if not event:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"System event with ID {event_id} not found"
|
|
)
|
|
return event
|
|
|
|
|
|
@router.post("", response_model=SystemEventSchema, status_code=status.HTTP_201_CREATED)
|
|
async def create_system_event(
|
|
event_in: SystemEventCreate,
|
|
db: Session = Depends(get_db)
|
|
) -> Any:
|
|
"""
|
|
Create a new system event.
|
|
"""
|
|
# Validate event type
|
|
valid_event_types = ["startup", "shutdown", "error", "warning", "info"]
|
|
if event_in.event_type not in valid_event_types:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid event type. Must be one of: {', '.join(valid_event_types)}"
|
|
)
|
|
|
|
# Create the system event
|
|
db_event = SystemEvent(
|
|
event_type=event_in.event_type,
|
|
component=event_in.component,
|
|
message=event_in.message,
|
|
details=event_in.details
|
|
)
|
|
|
|
db.add(db_event)
|
|
db.commit()
|
|
db.refresh(db_event)
|
|
|
|
return db_event |