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.
Load this skill when:
---
```
ktrdr/api/
├── endpoints/ # Route handlers
├── models/ # Pydantic request/response models
├── services/ # Business logic
└── main.py # Router registration
```
1. **Create endpoint** in `ktrdr/api/endpoints/`
2. **Define Pydantic models** in `ktrdr/api/models/`
3. **Implement business logic** in `ktrdr/api/services/`
4. **Register router** in `ktrdr/api/main.py`
5. **Add tests** in `tests/api/`
---
For long-running tasks (training, backtesting, data downloads):
```python
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:
```python
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:
```python
@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."""
```
---
```python
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"
}
}
```
```python
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:
```python
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:
---
```python
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()
```
---
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/api-development/raw