new_router_template.py•6.11 kB
"""
Template for creating new FastAPI routers.
Copy this file and modify it to create new routers quickly.
"""
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from maverick_mcp.data.models import get_db
from maverick_mcp.utils.logging import get_logger
logger = get_logger(__name__)
# Create router instance
router = APIRouter(
prefix="/your_domain", # Change this to your domain
tags=["your_domain"], # Change this to your domain
)
# Request/Response models
class YourRequest(BaseModel):
"""Request model for your endpoint."""
field1: str = Field(..., description="Description of field1")
field2: int = Field(10, ge=0, description="Description of field2")
field3: bool = Field(True, description="Description of field3")
class YourResponse(BaseModel):
"""Response model for your endpoint."""
status: str = Field(..., description="Operation status")
result: dict[str, Any] = Field(..., description="Operation result")
message: str | None = Field(None, description="Optional message")
# Endpoints
@router.get("/", response_model=list[YourResponse])
async def list_items(
limit: int = Query(10, ge=1, le=100, description="Number of items to return"),
offset: int = Query(0, ge=0, description="Number of items to skip"),
db: Session = Depends(get_db),
):
"""
List items with pagination.
Args:
limit: Maximum number of items to return
offset: Number of items to skip
db: Database session
Returns:
List of items
"""
try:
# Your business logic here
items = [] # Fetch from database
logger.info(
f"Listed {len(items)} items",
extra={"limit": limit, "offset": offset},
)
return items
except Exception as e:
logger.error(f"Error listing items: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/{item_id}", response_model=YourResponse)
async def get_item(
item_id: int,
db: Session = Depends(get_db),
):
"""
Get a specific item by ID.
Args:
item_id: The item ID
db: Database session
Returns:
The requested item
"""
try:
# Your business logic here
item = None # Fetch from database
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return YourResponse(
status="success",
result={"id": item_id, "data": "example"},
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error getting item {item_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.post("/", response_model=YourResponse)
async def create_item(
request: YourRequest,
db: Session = Depends(get_db),
):
"""
Create a new item.
Args:
request: The item data
db: Database session
Returns:
The created item
"""
try:
logger.info(
"Creating new item",
extra={"request": request.model_dump()},
)
# Your business logic here
# Example: Create in database
# new_item = YourModel(**request.model_dump())
# db.add(new_item)
# db.commit()
return YourResponse(
status="success",
result={"id": 1, "created": True},
message="Item created successfully",
)
except Exception as e:
logger.error(f"Error creating item: {e}", exc_info=True)
db.rollback()
raise HTTPException(status_code=500, detail="Internal server error")
@router.put("/{item_id}", response_model=YourResponse)
async def update_item(
item_id: int,
request: YourRequest,
db: Session = Depends(get_db),
):
"""
Update an existing item.
Args:
item_id: The item ID
request: The updated data
db: Database session
Returns:
The updated item
"""
try:
# Your business logic here
# Example: Update in database
# item = db.query(YourModel).filter(YourModel.id == item_id).first()
# if not item:
# raise HTTPException(status_code=404, detail="Item not found")
# Update fields
# for key, value in request.model_dump().items():
# setattr(item, key, value)
# db.commit()
return YourResponse(
status="success",
result={"id": item_id, "updated": True},
message="Item updated successfully",
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error updating item {item_id}: {e}", exc_info=True)
db.rollback()
raise HTTPException(status_code=500, detail="Internal server error")
@router.delete("/{item_id}")
async def delete_item(
item_id: int,
db: Session = Depends(get_db),
):
"""
Delete an item.
Args:
item_id: The item ID
db: Database session
Returns:
Deletion confirmation
"""
try:
# Your business logic here
# Example: Delete from database
# item = db.query(YourModel).filter(YourModel.id == item_id).first()
# if not item:
# raise HTTPException(status_code=404, detail="Item not found")
# db.delete(item)
# db.commit()
return {"status": "success", "message": f"Item {item_id} deleted"}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error deleting item {item_id}: {e}", exc_info=True)
db.rollback()
raise HTTPException(status_code=500, detail="Internal server error")
# Health check endpoint
@router.get("/health")
async def health_check():
"""Check if the router is healthy."""
return {
"status": "healthy",
"router": "your_domain",
"timestamp": "2024-01-01T00:00:00Z",
}