Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 18 additions & 17 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -51,23 +51,24 @@ repos:
rev: "6.0.1"
hooks:
- id: isort
- repo: https://github.com/dosisod/refurb
rev: v2.0.0
hooks:
- id: refurb
additional_dependencies:
- boto3
- django-constance
- django-cors-headers
- django-environ
- django-extensions
- django-filter
- django-simple-history
- django-stubs[compatible-mypy]
- drf-spectacular
- drf-standardized-errors
- djangorestframework-stubs[compatible-mypy]
- zappa-django-utils
# TODO: https://github.com/dosisod/refurb/issues/372 문제 해소 후 uncomment 필요
# - repo: https://github.com/dosisod/refurb
# rev: v2.0.0
# hooks:
# - id: refurb
# additional_dependencies:
# - boto3
# - django-constance
# - django-cors-headers
# - django-environ
# - django-extensions
# - django-filter
# - django-simple-history
# - django-stubs[compatible-mypy]
# - drf-spectacular
# - drf-standardized-errors
# - djangorestframework-stubs[compatible-mypy]
# - zappa-django-utils
- repo: https://github.com/astral-sh/uv-pre-commit
rev: 0.6.12
hooks:
Expand Down
88 changes: 70 additions & 18 deletions app/core/viewset/json_schema_viewset.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from __future__ import annotations

import functools
import typing

from core.const.tag import OpenAPITag
Expand All @@ -23,26 +22,74 @@ def __new__(cls, *args: tuple, **kwargs: dict) -> JsonSchemaViewSet:
return super().__new__(cls)

@staticmethod
@functools.lru_cache
def get_enum_values(model_qs: QuerySet, is_nullable: bool) -> list[dict[str, str]]:
enum_values: list[dict[str, str]] = [{"const": None, "title": "빈 값"}] if is_nullable else []

qs = model_qs.all()
if hasattr(qs, "filter_active"):
qs = qs.filter_active()
elif hasattr(model_qs.model, "is_active"):
qs = qs.filter(is_active=True)
def _get_choices_from_queryset(qs: QuerySet, is_nullable: bool) -> list[dict[str, str]]:
choices: list[dict[str, str]] = [{"const": None, "title": "빈 값"}] if is_nullable else []

related_model = qs.model
if hasattr(related_model, "get_choices_queryset"):
qs = related_model.get_choices_queryset()
else:
qs = qs.all()
if hasattr(qs, "filter_active"):
qs = qs.filter_active()
elif hasattr(related_model, "is_active"):
qs = qs.filter(is_active=True)

for row in qs:
enum_values.append({"const": str(row.pk), "title": str(row)})
choices.append({"const": str(row.pk), "title": str(row)})

return enum_values
return choices

@staticmethod
def set_ui_schema(ui_schema: dict, field_name: str, data: dict) -> None:
ui_schema.setdefault(field_name, {})
ui_schema[field_name].update(data)

def _get_related_field_info(self) -> list[tuple[str, object, serializers.Field, bool]]:
"""Returns list of (field_name, model_field, serializer_field, is_m2m) for FK/M2M fields."""
serializer_class = typing.cast(type[JsonSchemaSerializer], self.get_serializer_class())

if not hasattr(serializer_class.Meta, "model"):
return []

ser_fields: dict[str, serializers.Field] = serializer_class().fields
model_fields = serializer_class.Meta.model._meta.fields
model_m2m_fields = serializer_class.Meta.model._meta.many_to_many
schema = serializer_class.get_json_schema()

result = []
for field in model_fields + model_m2m_fields:
if field.name not in schema.get("properties", {}) or field.name not in ser_fields:
continue

serializer_field = ser_fields[field.name]

if isinstance(field, ForeignKey):
s_field = typing.cast(serializers.PrimaryKeyRelatedField | None, serializer_field)
if not s_field or serializer_field.read_only:
continue
result.append((field.name, field, serializer_field, False))
elif isinstance(field, ManyToManyField):
s_field = typing.cast(serializers.ManyRelatedField | None, serializer_field)
if not s_field or serializer_field.read_only:
continue
result.append((field.name, field, serializer_field, True))

return result

def get_choices(self) -> dict[str, list[dict[str, str]]]:
choices: dict[str, list[dict[str, str]]] = {}

for field_name, field, serializer_field, is_m2m in self._get_related_field_info():
if is_m2m:
qs = typing.cast(serializers.ManyRelatedField, serializer_field).child_relation.get_queryset()
choices[field_name] = self._get_choices_from_queryset(qs, False)
else:
qs = typing.cast(serializers.PrimaryKeyRelatedField, serializer_field).get_queryset()
choices[field_name] = self._get_choices_from_queryset(qs, field.null)

return choices

def get_json_schema(self) -> dict: # noqa: C901
serializer_class = typing.cast(type[JsonSchemaSerializer], self.get_serializer_class())

Expand Down Expand Up @@ -70,19 +117,15 @@ def get_json_schema(self) -> dict: # noqa: C901
serializer_field = ser_fields[field.name]

if isinstance(field, ForeignKey):
if not (s_field := typing.cast(serializers.PrimaryKeyRelatedField | None, serializer_field)):
if not typing.cast(serializers.PrimaryKeyRelatedField | None, serializer_field):
continue
if serializer_field.read_only:
continue
e_values = self.get_enum_values(s_field.get_queryset(), field.null)
result["schema"]["properties"][field.name]["oneOf"] = e_values
elif isinstance(field, ManyToManyField):
if not (s_field := typing.cast(serializers.ManyRelatedField | None, serializer_field)):
if not typing.cast(serializers.ManyRelatedField | None, serializer_field):
continue
if serializer_field.read_only:
continue
e_values = self.get_enum_values(s_field.child_relation.get_queryset(), False)
result["schema"]["properties"][field.name]["items"]["oneOf"] = e_values
result["schema"]["properties"][field.name]["uniqueItems"] = True
self.set_ui_schema(result["ui_schema"], field.name, {"ui:field": "m2m_select"})
elif isinstance(field, FileField):
Expand Down Expand Up @@ -115,3 +158,12 @@ def get_json_schema(self) -> dict: # noqa: C901
@decorators.action(detail=False, methods=["get"], url_path="json-schema")
def response_json_schema(self, *args: tuple, **kwargs: dict) -> response.Response:
return response.Response(data=self.get_json_schema())

@utils.extend_schema(
tags=[OpenAPITag.ADMIN_JSON_SCHEMA],
summary="Choices for related fields",
responses={status.HTTP_200_OK: openapi.OpenApiResponse(response=types.OpenApiTypes.OBJECT)},
)
@decorators.action(detail=False, methods=["get"], url_path="choices")
def response_choices(self, *args: tuple, **kwargs: dict) -> response.Response:
return response.Response(data=self.get_choices())
Loading