認証機構の実装

APIリクエストの認証処理を実装します。

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from .models import UserToken

class ApiKeyAuth(BaseAuthentication):
    def authenticate(self, request):
        api_key = request.META.get('HTTP_X_API_KEY')
        token = UserToken.objects.filter(key=api_key).first()
        if token:
            return (token.user, api_key)
        raise AuthenticationFailed('無効な認証情報です')

グローバル設定（settings.py）

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'your_app.auth.ApiKeyAuth',
    ]
}

ローカル設定（views.py）

from .auth import ApiKeyAuth

class DataView(APIView):
    authentication_classes = [ApiKeyAuth]

権限制御の実装

特定のユーザー権限をチェックします。

from rest_framework.permissions import BasePermission

class AdminPermission(BasePermission):
    def has_permission(self, request, view):
        if request.user.role == 'admin':
            return True
        self.message = '管理者権限が必要です'
        return False

グローバル設定（settings.py）

REST_FRAMEWORK = {
    'DEFAULT_PERMISSION_CLASSES': [
        'your_app.permissions.AdminPermission',
    ],
}

リクエスト制限の実装

リクエスト頻度制限を設定します。

from rest_framework.throttling import SimpleRateThrottle

class RequestThrottle(SimpleRateThrottle):
    rate = '5/min'
    
    def get_cache_key(self, request, view):
        return f'ip:{request.META.get("REMOTE_ADDR")}'

グローバル設定（settings.py）

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'your_app.throttles.RequestThrottle',
    ],
}

カスタム制限クラスの実装例

from rest_framework.throttling import BaseThrottle
from django.core.cache import cache
import time

class CustomRateLimiter(BaseThrottle):
    cache_key_prefix = 'rate_limit:'
    rate = '10/min'
    
    def __init__(self):
        self.num, self.period = self.parse_rate(self.rate)
    
    def parse_rate(self, rate):
        count, unit = rate.split('/')
        return int(count), 60 if unit == 'min' else 1
    
    def allow_request(self, request, view):
        ip = request.META.get('REMOTE_ADDR')
        cache_key = f'{self.cache_key_prefix}{ip}'
        access_times = cache.get(cache_key, [])
        
        # 有効期限内のアクセス履歴をフィルタリング
        current_time = time.time()
        access_times = [t for t in access_times if current_time - t < self.period]
        
        if len(access_times) >= self.num:
            wait_time = self.period - (current_time - access_times[-1])
            raise Throttled(f'制限超過。{int(wait_time)}秒後に再試行してください')
        
        access_times.append(current_time)
        cache.set(cache_key, access_times, timeout=self.period)
        return True