On this page
FastAPI Endpoint
"""
FastAPI Endpoint Template
This is a complete, production-ready template for FastAPI endpoints.
Copy this file and modify for your use case.
Key components:
- Router setup with prefix and tags
- Pydantic request/response models
- Dependency injection (database session, current user)
- Service layer call for business logic
- Comprehensive error handling (401, 403, 422, 400)
- Structured logging
- Complete docstrings (Google style)
"""
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field
from app.api import deps
from app.models.user import User
from app.services.example_service import ExampleService
from app.core.logging import get_logger
# Initialize logger
logger = get_logger(__name__)
# Create router
router = APIRouter()
# ==============================================================================
# PYDANTIC SCHEMAS (Request/Response Models)
# ==============================================================================
class ExampleItemCreate(BaseModel):
"""
Schema for creating a new example item.
Attributes:
name: Name of the item (1-100 characters)
description: Optional description
price: Item price (must be positive)
is_active: Whether item is active (default: True)
"""
name: str = Field(..., min_length=1, max_length=100, description="Item name")
description: Optional[str] = Field(None, max_length=500, description="Item description")
price: float = Field(..., gt=0, description="Item price (must be positive)")
is_active: bool = Field(True, description="Whether item is active")
class ExampleItemUpdate(BaseModel):
"""
Schema for updating an existing example item.
All fields are optional - only provided fields will be updated.
"""
name: Optional[str] = Field(None, min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: Optional[float] = Field(None, gt=0)
is_active: Optional[bool] = None
class ExampleItemResponse(BaseModel):
"""
Schema for example item response.
This is what clients receive when fetching items.
"""
id: int
name: str
description: Optional[str]
price: float
is_active: bool
user_id: int
created_at: str
class Config:
from_attributes = True # Allows loading from SQLAlchemy models
# ==============================================================================
# ENDPOINTS
# ==============================================================================
@router.post(
"/items",
response_model=ExampleItemResponse,
status_code=status.HTTP_201_CREATED,
summary="Create a new item",
tags=["items"]
)
def create_item(
item_in: ExampleItemCreate,
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_user),
) -> ExampleItemResponse:
"""
Create a new item.
This endpoint creates a new item owned by the authenticated user.
Args:
item_in: Item creation data (name, description, price)
db: Database session (automatically injected)
current_user: Currently authenticated user (automatically injected)
Returns:
ExampleItemResponse: The newly created item
Raises:
HTTPException 401: If user is not authenticated
HTTPException 403: If user is not active
HTTPException 422: If validation fails (handled automatically by Pydantic)
HTTPException 400: If business logic validation fails
Example:
```python
POST /api/v1/items
Headers:
Authorization: Bearer <token>
Body:
{
"name": "Product A",
"description": "A great product",
"price": 29.99,
"is_active": true
}
Response (201):
{
"id": 123,
"name": "Product A",
"description": "A great product",
"price": 29.99,
"is_active": true,
"user_id": 456,
"created_at": "2025-01-15T10:30:00Z"
}
```
"""
logger.info(f"Creating item for user {current_user.id}: {item_in.name}")
# Authorisation check example
if not current_user.is_active:
logger.warning(f"Inactive user {current_user.id} attempted to create item")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Your account is inactive. Please contact support."
)
# Call service layer for business logic
try:
item = ExampleService.create_item(db, current_user.id, item_in)
logger.info(f"Item {item.id} created successfully by user {current_user.id}")
return item
except ValueError as e:
# Service layer raises ValueError for business logic errors
logger.error(f"Failed to create item: {str(e)}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
@router.get(
"/items",
response_model=List[ExampleItemResponse],
summary="List all items",
tags=["items"]
)
def list_items(
skip: int = Query(0, ge=0, description="Number of items to skip"),
limit: int = Query(20, ge=1, le=100, description="Max items to return"),
is_active: Optional[bool] = Query(None, description="Filter by active status"),
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_user),
) -> List[ExampleItemResponse]:
"""
List items with pagination and filtering.
Retrieves items owned by the authenticated user with optional filtering.
Args:
skip: Number of items to skip (for pagination)
limit: Maximum number of items to return (1-100)
is_active: Optional filter by active status
db: Database session (automatically injected)
current_user: Currently authenticated user (automatically injected)
Returns:
List[ExampleItemResponse]: List of items
Raises:
HTTPException 401: If user is not authenticated
Example:
```python
GET /api/v1/items?skip=0&limit=20&is_active=true
Headers:
Authorization: Bearer <token>
Response (200):
[
{
"id": 123,
"name": "Product A",
"description": "A great product",
"price": 29.99,
"is_active": true,
"user_id": 456,
"created_at": "2025-01-15T10:30:00Z"
},
...
]
```
"""
logger.info(
f"Listing items for user {current_user.id} "
f"(skip={skip}, limit={limit}, is_active={is_active})"
)
items = ExampleService.list_items(
db,
user_id=current_user.id,
skip=skip,
limit=limit,
is_active=is_active
)
logger.info(f"Returned {len(items)} items for user {current_user.id}")
return items
@router.get(
"/items/{item_id}",
response_model=ExampleItemResponse,
summary="Get item by ID",
tags=["items"]
)
def get_item(
item_id: int,
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_user),
) -> ExampleItemResponse:
"""
Retrieve a specific item by ID.
Args:
item_id: ID of the item to retrieve
db: Database session (automatically injected)
current_user: Currently authenticated user (automatically injected)
Returns:
ExampleItemResponse: The requested item
Raises:
HTTPException 401: If user is not authenticated
HTTPException 403: If user doesn't own the item
HTTPException 404: If item not found
Example:
```python
GET /api/v1/items/123
Headers:
Authorization: Bearer <token>
Response (200):
{
"id": 123,
"name": "Product A",
"price": 29.99,
...
}
```
"""
logger.info(f"Fetching item {item_id} for user {current_user.id}")
try:
item = ExampleService.get_item(db, item_id, current_user.id)
if not item:
logger.warning(f"Item {item_id} not found or unauthorized")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item {item_id} not found"
)
logger.info(f"Item {item_id} retrieved successfully")
return item
except PermissionError as e:
# User doesn't own this item
logger.warning(f"User {current_user.id} unauthorized for item {item_id}")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to access this item"
)
@router.patch(
"/items/{item_id}",
response_model=ExampleItemResponse,
summary="Update item",
tags=["items"]
)
def update_item(
item_id: int,
item_in: ExampleItemUpdate,
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_user),
) -> ExampleItemResponse:
"""
Update an existing item.
Only provided fields will be updated. Omitted fields remain unchanged.
Args:
item_id: ID of the item to update
item_in: Fields to update
db: Database session (automatically injected)
current_user: Currently authenticated user (automatically injected)
Returns:
ExampleItemResponse: The updated item
Raises:
HTTPException 401: If user is not authenticated
HTTPException 403: If user doesn't own the item
HTTPException 404: If item not found
HTTPException 400: If validation fails
Example:
```python
PATCH /api/v1/items/123
Headers:
Authorization: Bearer <token>
Body:
{
"price": 39.99,
"is_active": false
}
Response (200):
{
"id": 123,
"name": "Product A", // Unchanged
"price": 39.99, // Updated
"is_active": false, // Updated
...
}
```
"""
logger.info(f"Updating item {item_id} for user {current_user.id}")
try:
item = ExampleService.update_item(db, item_id, current_user.id, item_in)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item {item_id} not found"
)
logger.info(f"Item {item_id} updated successfully")
return item
except PermissionError:
logger.warning(f"User {current_user.id} unauthorized to update item {item_id}")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to update this item"
)
except ValueError as e:
logger.error(f"Failed to update item {item_id}: {str(e)}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
@router.delete(
"/items/{item_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete item",
tags=["items"]
)
def delete_item(
item_id: int,
db: Session = Depends(deps.get_db),
current_user: User = Depends(deps.get_current_user),
) -> None:
"""
Delete an item (soft delete).
Marks the item as deleted rather than actually removing it from the database.
Args:
item_id: ID of the item to delete
db: Database session (automatically injected)
current_user: Currently authenticated user (automatically injected)
Returns:
None: 204 No Content on success
Raises:
HTTPException 401: If user is not authenticated
HTTPException 403: If user doesn't own the item
HTTPException 404: If item not found
Example:
```python
DELETE /api/v1/items/123
Headers:
Authorization: Bearer <token>
Response: 204 No Content
```
"""
logger.info(f"Deleting item {item_id} for user {current_user.id}")
try:
deleted = ExampleService.delete_item(db, item_id, current_user.id)
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item {item_id} not found"
)
logger.info(f"Item {item_id} deleted successfully")
return None
except PermissionError:
logger.warning(f"User {current_user.id} unauthorized to delete item {item_id}")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You don't have permission to delete this item"
)
# ==============================================================================
# HOW TO USE THIS TEMPLATE
# ==============================================================================
"""
1. Copy this file to app/api/v1/endpoints/your_endpoint.py
2. Replace "Example" with your domain name:
- ExampleItemCreate → YourItemCreate
- ExampleItemResponse → YourItemResponse
- ExampleService → YourService
3. Update the Pydantic schemas with your fields
4. Implement your service layer (app/services/your_service.py)
5. Update router prefix and tags
6. Register router in app/api/v1/__init__.py:
```python
from app.api.v1.endpoints import your_endpoint
api_router = APIRouter()
api_router.include_router(
your_endpoint.router,
prefix="/your-endpoint",
tags=["your-tag"]
)
```
7. Write tests in tests/test_api/test_your_endpoint.py
8. Update docs/api/endpoints.md with your new endpoints
"""