DRFソースコード解析-シリアライザ/リクエスト処理/レスポンス処理/例外処理/レンダリング/10種類のAPI実装

概要

Django REST Framework(DRF)を使用する際、フロントエンドとバックエンドは分離されています。データ交換にはJSONが主流で、バックエンドではPython辞書(dict)とJSONの相互変換が必要です。jsonモジュールのloads/dumpsで手動変換可能ですが、DRFのシリアライザ機能により、この処理を自動化できます。シリアライザはデータ検証やカスタムロジックの挿入ポイントも提供します。

DRFでは以下のようなシリアライザクラスが用意されています:

Serializer: 基底クラスで、すべてのフィールドとcreate/updateメソッドを手動で実装
ModelSerializer: モデルからフィールドとCRUDメソッドを自動生成
ListSerializer: 複数オブジェクトの処理をカスタマイズ（例: バッチ更新）
HyperlinkedModelSerializer: 関連モデルをURLで表現

Serializerクラスの定義

from rest_framework import serializers
from app.models import Comment

class CommentSerializer(serializers.Serializer):
    email = serializers.EmailField()
    content = serializers.CharField(max_length=200)
    created = serializers.DateTimeField()

    def create(self, validated_data):
        return Comment(**validated_data)

    def update(self, instance, validated_data):
        instance.email = validated_data.get('email', instance.email)
        instance.content = validated_data.get('content', instance.content)
        instance.created = validated_data.get('created', instance.created)
        return instance

ModelSerializerとListSerializer

from rest_framework.serializers import ModelSerializer, ListSerializer
from rest_framework.exceptions import ValidationError
from books.models import Book, Author, AuthorDetail, Publish

class V2BookListSerializer(ListSerializer):
    def update(self, instance, validated_data):
        for index, obj in enumerate(instance):
            self.child.update(instance=obj, validated_data=validated_data[index])
        return instance


class BookDataSerializer(ModelSerializer):
    class Meta:
        model = Book
        fields = ['title', 'price', 'cover', 'author_list', 'publisher_name', 'publishers', 'authors']
        extra_kwargs = {
            'title': {
                'min_length': 1,
                'required': True,
                'error_messages': {
                    'required': '必須項目です'
                }
            },
            'publishers': {
                'write_only': True
            },
            'authors': {
                'write_only': True
            },
            'cover': {
                'read_only': True
            }
        }
        list_serializer_class = V2BookListSerializer

    def validate_title(self, value):
        if len(value) > 15:
            raise ValidationError('15文字以内で入力してください')
        return value

    def validate(self, attrs):
        request_method = self.context.get('request').method
        if Book.objects.filter(title=attrs['title'], publishers=attrs['publishers']).exists():
            raise ValidationError({'status': 1, 'msg': '同じタイトルの本が既に存在します'})
        return attrs

Metaクラスの設定

model属性でモデルクラスを指定
fieldsリストにシリアライズ/デシリアライズするフィールドを記載
extra_kwargsで各フィールドの追加設定（例: read_only/write_only）
depth属性で関連モデルのネスト表示レベルを指定
群更新時はListSerializerをカスタマイズ

検証ロジックの実装

class BookDataSerializer(ModelSerializer):
    # 単一フィールド検証
    def validate_title(self, value):
        if len(value) > 15:
            raise ValidationError('15文字以内で入力してください')
        return value

    # 全体検証
    def validate(self, attrs):
        if Book.objects.filter(title=attrs['title'], publishers=attrs['publishers']).exists():
            raise ValidationError('既に同じタイトルの本が存在します')
        return attrs

Viewでの使用例

from rest_framework.views import APIView
from rest_framework.response import Response
from books.serializers import BookDataSerializer
from books.models import Book

class BookView(APIView):
    def get(self, request, *args, **kwargs):
        pk = kwargs.get('pk')
        if pk:
            try:
                book = Book.objects.get(pk=pk)
            except Book.DoesNotExist:
                return MyResponse(status=1, msg='該当する本が存在しません')
            serializer = BookDataSerializer(book)
        else:
            books = Book.objects.all()
            serializer = BookDataSerializer(books, many=True)
        return Response(serializer.data)

    def post(self, request, *args, **kwargs):
        serializer = BookDataSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data)

ソースコード解析

シリアライザの継承関係

print(BookDataSerializer.__mro__)
# 出力結果
(<class 'books.serializers.BookDataSerializer'>,
 <class 'rest_framework.serializers.ModelSerializer'>,
 <class 'rest_framework.serializers.Serializer'>,
 <class 'rest_framework.serializers.BaseSerializer'>,
 <class 'rest_framework.fields.Field'>,
 <class 'object'>)

インスタンス作成

def __init__(self, instance=None, data=empty, **kwargs):
    self.instance = instance
    if data is not empty:
        self.initial_data = data
    self.partial = kwargs.pop('partial', False)
    self._context = kwargs.pop('context', {})

データ検証

def is_valid(self, raise_exception=False):
    if not hasattr(self, '_validated_data'):
        try:
            self._validated_data = self.run_validation(self.initial_data)
        except ValidationError as exc:
            self._errors = exc.detail
    if self._errors and raise_exception:
        raise ValidationError(self.errors)
    return not bool(self._errors)

保存処理

def save(self, **kwargs):
    if self.instance is not None:
        self.instance = self.update(self.instance, validated_data)
    else:
        self.instance = self.create(validated_data)
    return self.instance

リクエスト処理

def initialize_request(self, request, *args, **kwargs):
    return Request(
        request,
        parsers=self.get_parsers(),
        authenticators=self.get_authenticators()
    )

レスポンス処理

class CustomResponse(Response):
    def __init__(self, status=0, msg='成功', **kwargs):
        data = {
            'status': status,
            'message': msg,
            **kwargs
        }
        super().__init__(data=data)

レンダリング設定

REST_FRAMEWORK = {
    'DEFAULT_RENDERER_CLASSES': [
        'rest_framework.renderers.JSONRenderer',
        'rest_framework.renderers.BrowsableAPIRenderer'
    ],
    'EXCEPTION_HANDLER': 'utils.exceptions.custom_exception_handler'
}

例外処理

def custom_exception_handler(exc, context):
    response = drf_exception_handler(exc, context)
    if response is None:
        return Response({'error': '内部サーバーエラー'}, status=500)
    return response

10種類のAPI実装

urlpatterns = [
    path('books/', BookView.as_view()),
    path('books/<int:pk>/', BookView.as_view()),
]

class BookView(APIView):
    def get(self, request, pk=None):
        if pk:
            try:
                book = Book.objects.get(pk=pk)
            except Book.DoesNotExist:
                return MyResponse(status=1, msg='本が見つかりません')
            return MyResponse(result=BookDataSerializer(book).data)
        else:
            books = Book.objects.all()
            return MyResponse(result=BookDataSerializer(books, many=True).data)

    def post(self, request):
        if isinstance(request.data, dict):
            serializer = BookDataSerializer(data=request.data)
        elif isinstance(request.data, list):
            serializer = BookDataSerializer(data=request.data, many=True)
        else:
            return MyResponse(status=1, msg='無効なリクエスト形式')
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return MyResponse(result=serializer.data)

    def delete(self, request, pk=None):
        if pk:
            Book.objects.filter(pk=pk).update(is_deleted=True)
        else:
            Book.objects.filter(pk__in=request.data['ids']).update(is_deleted=True)
        return MyResponse()

    def put(self, request, pk=None):
        if pk:
            book = Book.objects.get(pk=pk)
            serializer = BookDataSerializer(book, data=request.data)
        else:
            serializer = BookDataSerializer(data=request.data, many=True)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return MyResponse(result=serializer.data)

    def patch(self, request, pk=None):
        return self.put(request, pk, partial=True)

タグ: Django REST Framework シリアライザリクエスト処理レスポンス処理例外処理

5月22日 14:00 投稿

異端開発室