[Django] drf-spectacular를 이용한 API 문서 자동화(With DTO)

송희

20 min readJul 6, 2023

사내에 도입한 API 자동 문서화에 대한 내용들을 정리했습니다.

기존엔..

API 자동 문서를 도입하기 전에는 다음의 예시처럼 API 문서들을 하나하나 만들어서 작성했습니다.

처음에는 만드는 기능들이 많지 않아서 큰 불편함을 느끼지 못했습니다.

하지만 시간이 흐르면 흐를수록, 작업량이 많아지면 많아질수록

“이걸 손으로 하나하나 치는게 맞나?”

라는 의문이 들었습니다.

주변에 자문을 구하니, API 자동 문서화을 적용하고 있지 않은 회사가 없었습니다(..그걸 왜 손으로 치고 있냐며 혼났습니다..)

그래서 리드님께 상황에 대해서 여쭤보다보니, 결국 회사에 도입하는 역할을 맡게 되었습니다.

초기 도입기와 이슈들.

사실 처음에 django API 자동화문서를 리서치할 때까지만 해도 꿈과 희망에 젖어있었습니다.

리드님께서 말씀 주신 라이브러리는 drf-yasg 였지만,
여러 문제점들을 해결해준 drf-spectacular 를 접하게 되면서 이를 도입하면 문제가 없을 것이라고 생각했었습니다.

Django Rest Framework API Document Generator (feat. drf-spectacular)

drf-yasg의 단점 그리고 Open API Specification 3.0

techblog.yogiyo.co.kr

크게 어려울 것이 없다고 생각한 저는 공식문서를 참고하여 바로 이를 도입하였습니다.

되긴 되는데.. 큰 문제들이 있다

정의한 문제 상황은 다음과 같습니다.

- drf-spectacular에서는 request 및 response를 serializer의 schema를 따라서 자동으로 설정해준다

우리는 response를 serializer로 처리하지 않음

- drf 및 drf-spectacular에서는 multiple serializer를 지원하지 않는다.

우리는 output_dto를 이용해 한 응답값에 여러개의 serializer data를 넣고 있다.

위의 큰 문제들을 해결하기 위해, 며칠동안 공식문서와 Github 이슈들을 다시 꼼꼼히 읽고 다음의 방식들을 이용해봤습니다.

inline_serializer 로 schema를 입력할 수 있음

(ideation) inline_serializer에 serializer를 넣으면 안될까?

(check) serializer를 만들어서 사용하는 방식이라 안됨

(check) inline_serializer 는 serializer 없이 일회성 데이터 직렬이 필요할 때 사용

여러 개의 serializer를 하나의 serializer로 묶어서 보내기

(ideation) drf-yasg에서는 어떻게 처리할까? 가능할까??

(check) drf-yasg에서도 안 됨

(ideation) 여러 serializer를 인수로 가지고 있는 하나의 serializer를 생성하면 될까?

(check) 된다.

serializer data와 pandas를 이용한 dict가 섞여있는 부분은 어떻게 처리할까?

output_dto = dict(
            order_day_diet_food_logs=order_day_diet_food_log_data, # serializer.data
            day_diet_record=day_diet_record_data, # serializer.data
            food_products_to_survey=food_products_to_survey_data, # serializer.data
            proper_nutrient_range=proper_nutrient_range # dict임

        )

→ 진짜 모르겠어요..

‘설마 이거 AutoSchema 건들여야 하는건가?
에이…설마 아니겠지ㅋㅋ ㅋㅋㅋ
그러면 나 Pycon 갈 수 있겠는걸?

`drf-spectacular custom하기`

결국은 마지막으로 해결되지 않은 문제를 가지고 리드님께 찾아갔고,

장난삼아 의심했던 AutoSchema를 생성하는 코드를 건들여야하는 부분을 확인하고 오게 됐습니다.

리드님과 함께 drf-spectcular의 schema 작동 방식을 파악하고,

어느 메소드를 custom해야 할 지 리드님과 breakpoint를 찍어가면서 확인한 결과, AutoSchema 의 _get_response_bodies 를 override해서 custom 하도록 하면 된다는 점을 파악할 수 있었습니다.

⭐️ drf-spectcular에서 CustomSchema 등록하기(공식문서)

DRF 설정에 해당 CustomSchema를 등록해주면 됩니다.

REST_FRAMEWORK = {
    # YOUR SETTINGS
    'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
}

❓ 왜 DRF에서 설정하는 것이죠?
DRF도 문서화 기능을 제공합니다..!( django-rest-swagger )
다만, 활발히 유지보수되지 않고 drf-spectacular 가 이를 확장하여 더욱 강력하고 유연한 스키마 생성 및 문서화 기능을 제공하기 때문에 이를 이용합니다.
API View와 Serializer를 분석하여 스키마를 생성해야 하기 때문에, DRF 설정에 넣어주는 이유도 있습니다.

_get_response_bodies 를 custom하면 되는 부분을 확인했으니, 이제 이 부분을 custom 해 볼 시간입니다.

기존 함수는 이런식으로 reponse의 type을 확인하고 있습니다.

def _get_response_bodies(self, direction='response'):
    response_serializers = self.get_response_serializers()

    if (
        is_serializer(response_serializers)
        or is_basic_type(response_serializers)
        or isinstance(response_serializers, OpenApiResponse)
    ):
        if self.method == 'DELETE':
            return {'204': {'description': _('No response body')}}
        if self._is_create_operation():
            return {'201': self._get_response_for_code(response_serializers, '201', direction=direction)}
        return {'200': self._get_response_for_code(response_serializers, '200', direction=direction)}
    elif isinstance(response_serializers, dict):
        # custom handling for overriding default return codes with @extend_schema
        responses = {}
        for code, serializer in response_serializers.items():
            if isinstance(code, tuple):
                code, media_types = str(code[0]), code[1:]
            else:
                code, media_types = str(code), None
            content_response = self._get_response_for_code(serializer, code, media_types, direction)
            if code in responses:
                responses[code]['content'].update(content_response['content'])
            else:
                responses[code] = content_response
        return responses
    else:
        warn(
            f'could not resolve "{response_serializers}" for {self.method} {self.path}. '
            f'Expected either a serializer or some supported override mechanism. '
            f'Defaulting to generic free-form object.'
        )
        schema = build_basic_type(OpenApiTypes.OBJECT)
        schema['description'] = _('Unspecified response body')
        return {'200': self._get_response_for_code(schema, '200', direction=direction)}

custom한 함수

dto_name이라는 attribute가 있으면 해당 부분을 resolve하도록 추가하였습니다.
(* 필자: resolve는 소스코드에서 serialzier를 처리하는 함수의 이름을 따왔습니다)


class DTOSchema(AutoSchema):

  def _get_response_bodies(self, direction='response'):
      response_serializers = self.get_response_serializers()
  
      if hasattr(response_serializers, 'dto_name'):
          response_serializers: BaseDTO
          component = self.resolve_dto(response_serializers)
          response = {200: {'content': {'application/json': {'schema': component.ref}}}}
          return response
      else:
          response = super()._get_response_bodies(direction)
          return response

   ...

  def resolve_dto(self, dto,) -> ResolvedComponent:
    component = ResolvedComponent(
        name=dto.dto_name,
        type=ResolvedComponent.SCHEMA,
        object=dto,
    )
    if component not in self.registry:
        dto.schema_render(self)

    return self.registry[component]

resolve_dto 메소드에서는

Auto Schema의 component를 만들고
만든 component를 registry에 저장하는 역할을 합니다.

❓ Registry란?
API 뷰와 관련된 정보들을 등록하고 유지하는 중앙 저장소의 역할
스키마를 생성하고 문서화하는 과정에서 참조할 수 있는 정보를 관리

사내에서 사용하는 DTO라는 패턴을, API 문서에 적용될 수 있도록 형식을 변화하기 위해서 다음과 같은 메소드를 추가로 작성하였습니다.

@classmethod
def schema_render(cls, auto_schema: Optional['DTOSchema'] = None):
    responses = {}
    component = ResolvedComponent(
        name=cls.dto_name,
        type=ResolvedComponent.SCHEMA,
        object=cls,
    )
    if component in auto_schema.registry:
        # 이거면 맨 처음이 아님
        return auto_schema.registry[component].ref
    for key, val in cls.dto_fields.items():
        responses[key] = build_type(val, auto_schema)
    component = ResolvedComponent(
        name=cls.dto_name,
        type=ResolvedComponent.SCHEMA,
        object=cls,
        schema={'type': 'object', 'properties': responses}
    )
    auto_schema.registry.register(component)
    return auto_schema.registry[component].ref



def build_type(schema, auto_schema: Optional['DTOSchema'] = None):
    openapi_type_mapping = get_openapi_type_mapping()
    origin_schema = typing.get_origin(schema)
    if hasattr(schema, 'dto_fields'):
        val: BaseDTO
        return schema.schema_render(auto_schema)
    elif origin_schema is Union:
        return build_union_type(schema, auto_schema)
    elif origin_schema is list:
        return build_array_type(schema, auto_schema)
    elif origin_schema is dict:
        return openapi_type_mapping[PYTHON_TYPE_MAPPING[schema.__origin__]]
    elif type(schema) is ForwardRef:
        return build_forward_value(schema, auto_schema)
    else:
        return openapi_type_mapping[PYTHON_TYPE_MAPPING[schema]]


def build_array_type(schema, auto_schema: Optional['DTOSchema'] = None):
    schema = typing.get_args(schema)[0]
    return {'type': 'array', 'items': build_type(schema, auto_schema)}


def build_union_type(schema, auto_schema: Optional['DTOSchema'] = None):
    responses = {}
    union_args: tuple = typing.get_args(schema)
    if type(None) in union_args:
        child_val = list(set(union_args) - {type(None)})[0]
        responses['required'] = False
    else:
        child_val = union_args[0]
    a = build_type(child_val, auto_schema)
    return {**responses, **a}


def build_forward_value(schema: ForwardRef, auto_schema: Optional['DTOSchema'] = None):
    try:
        _schema = schema.__forward_value__
        if _schema is None:
            _schema = _DTODict[schema.__forward_arg__]
        if _schema is None:
            raise Exception()
        return build_type(_schema, auto_schema)
    except:
        '''warning 아직 정의되지 않은 forward value입니다.'''
        return {'type': schema.__forward_arg__}

그래도 남아있던 일들 — Respone, Pagination

위까지의 과정을 통해서 schema 생성 작업이 끝났습니다!

정말 끝인 줄 알았지만.. 실제 api respose값들과 괴리감이 있어 맞추기 위한 작업들이 남아있었습니다.

사내에서는

특이사항이 없는 한 아래처럼 DTOResponseFormatter 를 이용하여 return type을 보장하고 있습니다. -> {"results": data} 형식

return Response(DTOResponseFormatter.run(output_dto))

if dto_results is not None:
    rtn['results'] = dto_results # 이런 형식입니다.

모든 api에 사용되는 범용적인 함수인만큼, 모든 api에 해당 부분을 넣기보다는 범용적으로 처리하는 함수를 만들어야겠다고 판단했습니다.

고민 사항을 토대로, 다음의 함수를 만들었습니다.

many 옵션을 이용하여 특정 data들이 List로 return 될 때 ( {"results" : [data]} ) 도 이용 가능하도록 하였습니다.

defDTOResponse(response_dto, many=None):
    @dataclass
    class Run(response_dto):
        if many:
            result_dto = list[response_dto]
        else:
            result_dto = response_dto

        results: result_dto

        dto_name: str = field(default=f'{response_dto.dto_name}_output_dto', init=False)

    return Run

Pagination 부분도 이와 같은 맥락으로 처리하였습니다.

def PaginationDTOResponse(response_dto, many=None):
    @dataclass
    class Run(response_dto):
        if many:
            result_dto = list[response_dto]
        else:
            result_dto = response_dto

        results: Optional[result_dto]
        count: int
        next: Optional[str]
        prev: Optional[str]

        dto_name: str = field(default=f'pagination_{response_dto.dto_name}_output_dto', init=False)

    return Run

이를 코드에 적용하면 다음과 같이 이용할 수 있습니다.

@extend_schema(
        responses=DTOResponse(FoodProductCurationListQsFoodProductOutputDTO, many=True),
        parameters=[
            OpenApiParameter(name='day', type=OpenApiTypes.INT, location=OpenApiParameter.QUERY)
        ]
    )

@extend_schema(
        request=FoodProductSearchSearchPostInputDTO, responses=PaginationDTOResponse(FoodProductSearchSearchOutputDTO),
        parameters=[
            OpenApiParameter(name='page', type=OpenApiTypes.INT, location=OpenApiParameter.QUERY),
            OpenApiParameter(name='page_size', type=OpenApiTypes.INT, location=OpenApiParameter.QUERY)
        ],
    )

배포 여정기

기능적인 작업들은 마무리했지만, 이 문서를 배포하는 일이 남아있었습니다.

API 문서의 취지를 고려해봤을때, 개발 서버 외에 QA, 운영 서버에 올리는 것은 맞지 않다고 생각했습니다.

개발 서버에서만 API 문서 접근이 방법에 대해서 고민한 결과

개발서버에서만 url 연결

=> 사내에서 dev/staging/prod branch를 이용하여 dev의 branch를 staging에 merge시키는 방식으로 하기 때문에 불가능

2. 특정 IP 주소만 가능하게

=> 개발 서버에서만 가능하진 않음 + 새로운 직원들이 들어 왔을 때 + 원격 근무지 ip 주소를 하나하나 다 추가해야 하므로 매우 번거로움

3. Middleware를 이용하여 처리

=> ???

4. Permission을 이용하여 처리

=? ???

선택한 옵션은 3,4번이었고

3번의 경우 Middleware의 동작 원리를 생각해봤을 때, 모든 API 요청마다 확인해야하므로 범용적이지 않을 것 같다는 판단이 들어

4번을 선택하였습니다.

이를 위해 os.environ.get을 이용하여 서버를 확인하여 dev 서버에서만 접근이 가능한 permission을 만들었고, 이를 drf-spectacularsetting에 적용하였습니다.

class IsDevel(permissions.BasePermission):
    def has_permission(self, request, view):
        if settings.ENV_MODE in ['local', 'devel']:
            return True
        return False

SPECTACULAR_SETTINGS = {
    'TITLE': '마이쉽단 API',
      ... 
    'DISABLE_ERRORS_AND_WARNINGS': True, # warning과 error 안뜨게 설정
    'SERVE_PERMISSIONS': ['core.permissions.IsDevel'],

그 결과

dev에서는 작동하고 QA 서버에서는 작동하지 않습니다.