Use when creating new API endpoints, working with async operations, implementing long-running tasks with progress tracking, adding Pydantic models, or designing request/response schemas.
Sign in to like and favorite skills
Load this skill when:
ktrdr/api/ ├── endpoints/ # Route handlers ├── models/ # Pydantic request/response models ├── services/ # Business logic └── main.py # Router registration
ktrdr/api/endpoints/ktrdr/api/models/ktrdr/api/services/ktrdr/api/main.pytests/api/For long-running tasks (training, backtesting, data downloads):
from ktrdr.api.services.operations_service import OperationsService @router.post("/long-operation") async def start_operation( background_tasks: BackgroundTasks, operations_service: OperationsService = Depends(get_operations_service) ): # Register operation operation_id = await operations_service.register_operation( operation_type=OperationType.TRAINING, description="Training model..." ) # Start background task background_tasks.add_task( run_operation, operation_id, operations_service ) return {"operation_id": operation_id}
Operations should report progress for long-running tasks:
async def run_operation(operation_id: str, ops_service: OperationsService): try: await ops_service.update_status(operation_id, OperationStatus.RUNNING) for i, step in enumerate(steps): # Do work await process_step(step) # Update progress await ops_service.update_progress( operation_id, percentage=(i + 1) / len(steps) * 100, phase=f"Processing step {i + 1}" ) await ops_service.update_status(operation_id, OperationStatus.COMPLETED) except Exception as e: await ops_service.update_status( operation_id, OperationStatus.FAILED, error=str(e) ) raise
Standard endpoints for operation management:
@router.get("/operations/{operation_id}") async def get_operation_status(operation_id: str): """Get current status of an operation.""" @router.get("/operations/{operation_id}/metrics") async def get_operation_metrics(operation_id: str): """Get detailed metrics for an operation.""" @router.delete("/operations/{operation_id}/cancel") async def cancel_operation(operation_id: str): """Request cancellation of a running operation."""
from pydantic import BaseModel, Field class TrainingRequest(BaseModel): strategy_path: str = Field(..., description="Path to strategy YAML") symbol: str = Field(..., description="Trading symbol") class Config: json_schema_extra = { "example": { "strategy_path": "config/strategies/example.yaml", "symbol": "AAPL" } }
class OperationResponse(BaseModel): operation_id: str status: OperationStatus progress: float = 0.0 phase: str | None = None error: str | None = None created_at: datetime updated_at: datetime
Use HTTPException for API errors:
from fastapi import HTTPException @router.get("/resource/{id}") async def get_resource(id: str): resource = await fetch_resource(id) if not resource: raise HTTPException( status_code=404, detail=f"Resource {id} not found" ) return resource
Once server is running:
import pytest from httpx import AsyncClient @pytest.mark.asyncio async def test_create_operation(client: AsyncClient): response = await client.post( "/api/v1/operations", json={"type": "training", "params": {...}} ) assert response.status_code == 200 assert "operation_id" in response.json()
ktrdr/api/main.pyktrdr/api/services/operations_service.pyktrdr/api/dependencies.py