16 KiB
16 KiB
| name | description | argument-hint | allowed-tools |
|---|---|---|---|
| lp-drf-api | Creates DRF 3.16 API endpoints for league-planner using function-based views, @api_view decorator, drf-spectacular OpenAPI docs, and custom auth. Use for new API endpoints. | <endpoint-name> [HTTP-methods...] | Read, Write, Edit, Glob, Grep |
League-Planner DRF API Generator
Creates REST API endpoints following league-planner patterns: function-based views with @api_view, drf-spectacular documentation, custom token authentication, and proper error handling.
When to Use
- Creating new API endpoints in
/api/uefa/,/api/court/,/api/configurator/,/api/collector/ - Adding OpenAPI documentation to existing endpoints
- Implementing custom authentication or permissions
- Building serializers for complex nested data
Prerequisites
- Endpoint belongs to an existing API namespace (uefa, court, configurator, collector)
- Models and relationships are already defined
- URL routing is set up in the appropriate
urls.py
Instructions
Step 1: Create Serializer
Place in api/{namespace}/serializers.py:
from rest_framework import serializers
from drf_spectacular.utils import extend_schema_serializer, OpenApiExample
from scheduler.models import Team, Season
@extend_schema_serializer(
examples=[
OpenApiExample(
'Team Response',
value={
'id': 1,
'name': 'FC Bayern München',
'country': {'code': 'DE', 'name': 'Germany'},
'stadium': 'Allianz Arena',
},
response_only=True,
),
]
)
class TeamSerializer(serializers.ModelSerializer):
"""Serializer for Team with nested country."""
country = serializers.SerializerMethodField()
class Meta:
model = Team
fields = ['id', 'name', 'country', 'stadium']
read_only_fields = ['id']
def get_country(self, obj) -> dict | None:
if obj.country:
return {
'code': obj.country.code,
'name': obj.country.name,
}
return None
class TeamCreateSerializer(serializers.ModelSerializer):
"""Separate serializer for creating teams."""
class Meta:
model = Team
fields = ['name', 'country', 'stadium', 'latitude', 'longitude']
def validate_name(self, value: str) -> str:
if Team.objects.filter(name__iexact=value).exists():
raise serializers.ValidationError('Team name already exists')
return value
# Request serializers (for documentation)
class SeasonRequestSerializer(serializers.Serializer):
"""Request body for season-based endpoints."""
season_id = serializers.IntegerField(help_text='ID of the season')
include_inactive = serializers.BooleanField(
default=False,
help_text='Include inactive items'
)
class PaginatedResponseSerializer(serializers.Serializer):
"""Base pagination response."""
count = serializers.IntegerField()
next = serializers.URLField(allow_null=True)
previous = serializers.URLField(allow_null=True)
results = serializers.ListField()
Step 2: Create View Function
Place in api/{namespace}/views.py:
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from drf_spectacular.utils import (
extend_schema,
OpenApiParameter,
OpenApiResponse,
OpenApiExample,
)
from drf_spectacular.types import OpenApiTypes
from scheduler.models import Team, Season
from .serializers import (
TeamSerializer,
TeamCreateSerializer,
SeasonRequestSerializer,
)
@extend_schema(
summary='List Teams',
description='''
Returns all teams for a given season.
**Authentication**: Requires valid API key or session.
**Query Parameters**:
- `season_id`: Required. The season to fetch teams for.
- `active_only`: Optional. Filter to active teams only.
''',
parameters=[
OpenApiParameter(
name='season_id',
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
required=True,
description='Season ID to fetch teams for',
),
OpenApiParameter(
name='active_only',
type=OpenApiTypes.BOOL,
location=OpenApiParameter.QUERY,
required=False,
default=False,
description='Filter to active teams only',
),
],
responses={
200: OpenApiResponse(
response=TeamSerializer(many=True),
description='List of teams',
examples=[
OpenApiExample(
'Success',
value=[
{'id': 1, 'name': 'FC Bayern', 'country': {'code': 'DE'}},
{'id': 2, 'name': 'Real Madrid', 'country': {'code': 'ES'}},
],
),
],
),
400: OpenApiResponse(description='Invalid parameters'),
404: OpenApiResponse(description='Season not found'),
},
tags=['teams'],
)
@api_view(['GET'])
@permission_classes([AllowAny])
def list_teams(request, version=None):
"""List all teams for a season."""
season_id = request.query_params.get('season_id')
if not season_id:
return Response(
{'error': 'season_id is required'},
status=status.HTTP_400_BAD_REQUEST
)
try:
season = Season.objects.get(pk=season_id)
except Season.DoesNotExist:
return Response(
{'error': 'Season not found'},
status=status.HTTP_404_NOT_FOUND
)
teams = Team.objects.filter(
season=season
).select_related(
'country'
).order_by('name')
active_only = request.query_params.get('active_only', '').lower() == 'true'
if active_only:
teams = teams.filter(is_active=True)
serializer = TeamSerializer(teams, many=True)
return Response(serializer.data)
@extend_schema(
summary='Create Team',
description='Create a new team in the specified season.',
request=TeamCreateSerializer,
responses={
201: OpenApiResponse(
response=TeamSerializer,
description='Team created successfully',
),
400: OpenApiResponse(description='Validation error'),
},
tags=['teams'],
)
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def create_team(request, version=None):
"""Create a new team."""
serializer = TeamCreateSerializer(data=request.data)
if not serializer.is_valid():
return Response(
{'error': 'Validation failed', 'details': serializer.errors},
status=status.HTTP_400_BAD_REQUEST
)
team = serializer.save()
return Response(
TeamSerializer(team).data,
status=status.HTTP_201_CREATED
)
@extend_schema(
summary='Get Team Details',
description='Retrieve detailed information about a specific team.',
responses={
200: TeamSerializer,
404: OpenApiResponse(description='Team not found'),
},
tags=['teams'],
)
@api_view(['GET'])
def get_team(request, team_id: int, version=None):
"""Get team details by ID."""
try:
team = Team.objects.select_related('country', 'season').get(pk=team_id)
except Team.DoesNotExist:
return Response(
{'error': 'Team not found'},
status=status.HTTP_404_NOT_FOUND
)
serializer = TeamSerializer(team)
return Response(serializer.data)
@extend_schema(
summary='Bulk Update Teams',
description='Update multiple teams in a single request.',
request=TeamSerializer(many=True),
responses={
200: OpenApiResponse(
description='Teams updated',
response={'type': 'object', 'properties': {'updated': {'type': 'integer'}}},
),
},
tags=['teams'],
)
@api_view(['PATCH'])
@permission_classes([IsAuthenticated])
def bulk_update_teams(request, version=None):
"""Bulk update teams."""
updated_count = 0
for team_data in request.data:
team_id = team_data.get('id')
if not team_id:
continue
try:
team = Team.objects.get(pk=team_id)
serializer = TeamSerializer(team, data=team_data, partial=True)
if serializer.is_valid():
serializer.save()
updated_count += 1
except Team.DoesNotExist:
continue
return Response({'updated': updated_count})
Step 3: Configure URLs
In api/{namespace}/urls.py:
from django.urls import path
from . import views
app_name = 'api-uefa'
urlpatterns = [
# v1 endpoints
path('v1/teams/', views.list_teams, name='teams-list'),
path('v1/teams/create/', views.create_team, name='teams-create'),
path('v1/teams/<int:team_id>/', views.get_team, name='teams-detail'),
path('v1/teams/bulk/', views.bulk_update_teams, name='teams-bulk'),
# v2 endpoints (can have different implementations)
path('v2/teams/', views.list_teams, name='teams-list-v2'),
]
Patterns & Best Practices
Custom Authentication (Project Pattern)
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from scheduler.models import Team
class TeamTokenAuthentication(BaseAuthentication):
"""Authenticate via Team.hashval token."""
def authenticate(self, request):
token = request.headers.get('X-Team-Token')
if not token:
return None
try:
team = Team.objects.get(hashval=token)
return (None, team) # (user, auth)
except Team.DoesNotExist:
raise AuthenticationFailed('Invalid team token')
# Usage in view
@api_view(['GET'])
@authentication_classes([TeamTokenAuthentication])
def team_data(request):
team = request.auth # The authenticated team
return Response({'team': team.name})
Error Response Format
# Standard error structure used in the project
def error_response(code: str, message: str, details: dict = None, status_code: int = 400):
"""Create standardized error response."""
response_data = {
'error': {
'code': code,
'message': message,
}
}
if details:
response_data['error']['details'] = details
return Response(response_data, status=status_code)
# Usage
return error_response(
code='VALIDATION_ERROR',
message='Invalid input data',
details={'season_id': ['This field is required']},
status_code=status.HTTP_400_BAD_REQUEST
)
Session-Based Access (Project Pattern)
@api_view(['GET'])
def session_protected_view(request):
"""View requiring session-based authorization."""
# Check session for team access
authorized_league = request.session.get('authorized_league')
if not authorized_league:
return Response(
{'error': 'Not authorized'},
status=status.HTTP_403_FORBIDDEN
)
# Check session for club access
club_id = request.session.get('club')
if club_id:
# Handle club-specific logic
pass
return Response({'status': 'authorized'})
Pagination Pattern
from rest_framework.pagination import PageNumberPagination
class StandardPagination(PageNumberPagination):
page_size = 25
page_size_query_param = 'page_size'
max_page_size = 100
@api_view(['GET'])
def paginated_list(request):
"""Manually paginate in function-based view."""
queryset = Team.objects.all().order_by('name')
paginator = StandardPagination()
page = paginator.paginate_queryset(queryset, request)
if page is not None:
serializer = TeamSerializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
serializer = TeamSerializer(queryset, many=True)
return Response(serializer.data)
Versioning Pattern
@api_view(['GET'])
def versioned_endpoint(request, version=None):
"""Handle different API versions."""
if version == 'v1':
# V1 response format
return Response({'data': 'v1 format'})
elif version == 'v2':
# V2 response format with additional fields
return Response({'data': 'v2 format', 'extra': 'field'})
else:
return Response(
{'error': f'Unknown version: {version}'},
status=status.HTTP_400_BAD_REQUEST
)
drf-spectacular Configuration
Settings (leagues/settings.py)
SPECTACULAR_SETTINGS = {
'TITLE': 'League Planner API',
'DESCRIPTION': 'API for sports league planning and optimization',
'VERSION': '2.0.0',
'SERVE_INCLUDE_SCHEMA': False,
# Schema generation
'SCHEMA_PATH_PREFIX': r'/api/',
'SCHEMA_PATH_PREFIX_TRIM': True,
# Component naming
'COMPONENT_SPLIT_REQUEST': True,
'COMPONENT_NO_READ_ONLY_REQUIRED': True,
# Tags
'TAGS': [
{'name': 'teams', 'description': 'Team management'},
{'name': 'seasons', 'description': 'Season operations'},
{'name': 'draws', 'description': 'Tournament draws'},
],
# Security schemes
'SECURITY': [
{'ApiKeyAuth': []},
{'SessionAuth': []},
],
'APPEND_COMPONENTS': {
'securitySchemes': {
'ApiKeyAuth': {
'type': 'apiKey',
'in': 'header',
'name': 'X-API-Key',
},
'SessionAuth': {
'type': 'apiKey',
'in': 'cookie',
'name': 'sessionid',
},
},
},
}
Examples
Example 1: Complex Nested Response
class DrawResponseSerializer(serializers.Serializer):
"""Complex nested response for draw data."""
draw_id = serializers.IntegerField()
name = serializers.CharField()
groups = serializers.SerializerMethodField()
constraints = serializers.SerializerMethodField()
def get_groups(self, obj):
return [
{
'name': g.name,
'teams': [
{'id': t.id, 'name': t.name, 'country': t.country.code}
for t in g.teams.select_related('country')
]
}
for g in obj.groups.prefetch_related('teams__country')
]
def get_constraints(self, obj):
return list(obj.constraints.values('type', 'team1_id', 'team2_id'))
@extend_schema(
summary='Get Draw Results',
responses={200: DrawResponseSerializer},
tags=['draws'],
)
@api_view(['GET'])
def get_draw(request, draw_id: int):
draw = Draw.objects.prefetch_related(
'groups__teams__country',
'constraints',
).get(pk=draw_id)
serializer = DrawResponseSerializer(draw)
return Response(serializer.data)
Example 2: File Upload Endpoint
from rest_framework.parsers import MultiPartParser, FormParser
@extend_schema(
summary='Upload Team Logo',
request={
'multipart/form-data': {
'type': 'object',
'properties': {
'logo': {'type': 'string', 'format': 'binary'},
},
},
},
responses={200: TeamSerializer},
tags=['teams'],
)
@api_view(['POST'])
@parser_classes([MultiPartParser, FormParser])
def upload_logo(request, team_id: int):
team = Team.objects.get(pk=team_id)
logo = request.FILES.get('logo')
if not logo:
return error_response('MISSING_FILE', 'Logo file is required')
team.logo = logo
team.save()
return Response(TeamSerializer(team).data)
Common Pitfalls
- Missing
versionparameter: Always includeversion=Nonein view signature for versioned APIs - No query optimization: Use
select_related/prefetch_relatedbefore serializing - Inconsistent error format: Use the standard error response structure
- Missing OpenAPI docs: Every endpoint needs
@extend_schemafor documentation - Hardcoded URLs: Use
reverse()for generating URLs in responses
Verification
- Check OpenAPI schema:
python manage.py spectacular --color --file schema.yml - View Swagger UI at
/api/schema/swagger-ui/ - Test endpoint with curl:
curl -X GET "http://localhost:8000/api/uefa/v2/teams/?season_id=1" \ -H "Accept: application/json"