feat: Django Allauth 소셜 계정 관리 API 추가

app/admin_api/filtersets/socialaccount.py

@@ -0,0 +1,64 @@
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialAccount, SocialApp
+from core.filter.multi_field import MultiFieldOrCharInFilter
+from django_filters import rest_framework as filters
+
+
+def _social_app_provider_choices() -> list[tuple[str, str]]:
+    """SocialApp 에 등록된 provider 만 허용. callable 로 두어 매 요청마다 최신 DB 상태를 반영."""
+    return [(p, p) for p in SocialApp.objects.values_list("provider", flat=True).distinct().order_by("provider")]
+
+
+class SocialAppProviderInFilter(filters.BaseInFilter, filters.ChoiceFilter):
+    """CSV 다중값 입력 + SocialApp.provider 화이트리스트 검증 + `__in` 매칭. 각 값을 단일 choice 로 검증.
+
+    `expose_callable_choices_in_schema` 는 `core.openapi.filter_extension.DjangoFilterExtension` 가
+    스키마 생성 시 callable choices 를 한 번 호출하도록 opt-in.
+    """
+
+    expose_callable_choices_in_schema = True
+
+
+class SocialAccountAdminFilterSet(filters.FilterSet):
+    """admin 운영자 검색. provider 는 SocialApp 에 등록된 값만 허용 (`?provider=google,kakao`)."""
+
+    provider = SocialAppProviderInFilter(field_name="provider", choices=_social_app_provider_choices)
+    uid = filters.CharFilter(field_name="uid", lookup_expr="icontains")
+    user_email = filters.CharFilter(field_name="user__email", lookup_expr="icontains")
+    user_username = filters.CharFilter(field_name="user__username", lookup_expr="icontains")
+
+    date_joined_after = filters.DateTimeFilter(field_name="date_joined", lookup_expr="gte")
+    date_joined_before = filters.DateTimeFilter(field_name="date_joined", lookup_expr="lte")
+    last_login_after = filters.DateTimeFilter(field_name="last_login", lookup_expr="gte")
+    last_login_before = filters.DateTimeFilter(field_name="last_login", lookup_expr="lte")
+
+    class Meta:
+        model = SocialAccount
+        fields = [
+            "user",
+            "provider",
+            "uid",
+            "user_email",
+            "user_username",
+            "date_joined_after",
+            "date_joined_before",
+            "last_login_after",
+            "last_login_before",
+        ]
+
+
+class EmailAddressAdminFilterSet(filters.FilterSet):
+    """`email` 은 EmailAddress.email 과 User.email 양쪽을 OR 매칭. CSV 다중값 지원."""
+
+    email = MultiFieldOrCharInFilter(field_names=["email", "user__email"], lookup_expr="icontains")
+    user_username = filters.CharFilter(field_name="user__username", lookup_expr="icontains")
+
+    class Meta:
+        model = EmailAddress
+        fields = [
+            "user",
+            "email",
+            "verified",
+            "primary",
+            "user_username",
+        ]

app/admin_api/serializers/socialaccount.py

@@ -0,0 +1,66 @@
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialAccount, SocialApp
+from core.serializer.json_schema_serializer import JsonSchemaSerializer
+from core.serializer.lowercased_email_field import LowercasedEmailField
+from core.serializer.nested_model_serializer import NestedModelSerializer
+from rest_framework import serializers
+
+
+class SocialAppAdminSerializer(JsonSchemaSerializer, serializers.ModelSerializer):
+    str_repr = serializers.CharField(source="__str__", read_only=True)
+
+    class Meta:
+        model = SocialApp
+        fields = ("id", "provider", "provider_id", "name", "client_id", "secret", "key", "settings", "str_repr")
+        read_only_fields = ("id",)
+
+
+class SocialAccountAdminSerializer(JsonSchemaSerializer, serializers.ModelSerializer):
+    str_repr = serializers.CharField(source="__str__", read_only=True)
+
+    class Meta:
+        model = SocialAccount
+        read_only_fields = fields = (
+            "id",
+            "user",
+            "provider",
+            "uid",
+            "last_login",
+            "date_joined",
+            "extra_data",
+            "str_repr",
+        )
+
+
+class EmailAddressAdminSerializer(JsonSchemaSerializer, serializers.ModelSerializer):
+    str_repr = serializers.CharField(source="__str__", read_only=True)
+    email = LowercasedEmailField()
+
+    class Meta:
+        model = EmailAddress
+        fields = ("id", "user", "email", "verified", "primary", "str_repr")
+        extra_kwargs = {"id": {"read_only": True}}
+
+
+class EmailAddressNestedAdminSerializer(JsonSchemaSerializer, NestedModelSerializer):
+    id = serializers.IntegerField(required=False, help_text="기존 EmailAddress 수정 시 PK 전달, 새로 추가 시 생략")
+    email = LowercasedEmailField()
+
+    class Meta:
+        model = EmailAddress
+        fields = ("id", "user", "email", "verified", "primary")
+        # user 는 NestedFieldSpec.parent_fk_name 으로 부모 인스턴스에서 주입되므로 입력 시 생략 가능.
+        extra_kwargs = {"user": {"required": False}}
+        # validators=[] — auto UniqueTogetherValidator(user, email) 가 user 누락 시 required 로 막음.
+        # DB unique constraint(account_emailaddress_user_id_email) 가 여전히 enforce.
+        validators: list = []
+
+
+class SocialAccountNestedAdminSerializer(JsonSchemaSerializer, NestedModelSerializer):
+    # delete-only nested — id 로 기존 row 매칭만 함. UserAdminSerializer.validate_social_accounts 가 정책 강제.
+    id = serializers.IntegerField(required=True)
+
+    class Meta:
+        model = SocialAccount
+        read_only_fields = ("provider", "uid", "last_login", "date_joined", "extra_data")
+        fields = ("id",) + read_only_fields

app/admin_api/serializers/user.py

@@ -1,18 +1,25 @@
 import functools
 import typing
 
+from admin_api.serializers.socialaccount import EmailAddressNestedAdminSerializer, SocialAccountNestedAdminSerializer
+from admin_api.services.socialaccount import delete_social_accounts_and_cleanup_user_emails
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialAccount
 from core.const.account import generate_random_password
 from core.const.serializer import COMMON_ADMIN_FIELDS
 from core.serializer.base_abstract_serializer import BaseAbstractSerializer
 from core.serializer.json_schema_serializer import JsonSchemaSerializer
+from core.serializer.nested_model_serializer import NestedFieldModelSerializer, NestedFieldSpec
 from core.serializer.read_only_serializer import ReadOnlyModelSerializer
 from rest_framework import serializers
 from user.models import UserExt
 from user.models.organization import Organization
 
 
-class UserAdminSerializer(JsonSchemaSerializer, serializers.ModelSerializer):
+class UserAdminSerializer(JsonSchemaSerializer, NestedFieldModelSerializer):
     str_repr = serializers.CharField(source="__str__", read_only=True)
+    email_addresses = EmailAddressNestedAdminSerializer(many=True, required=False, source="emailaddress_set")
+    social_accounts = SocialAccountNestedAdminSerializer(many=True, required=False, source="socialaccount_set")
 
     class Meta:
         model = UserExt
@@ -28,17 +35,75 @@ class Meta:
             "str_repr",
             "date_joined",
             "last_login",
+            "email_addresses",
+            "social_accounts",
         )
         extra_kwargs = {
             "id": {"read_only": True},
             "date_joined": {"read_only": True},
             "last_login": {"read_only": True},
         }
+        nested_fields = {
+            "emailaddress_set": NestedFieldSpec(
+                related_manager_name="emailaddress_set",
+                child_model=EmailAddress,
+                parent_fk_name="user",
+            ),
+            "socialaccount_set": NestedFieldSpec(
+                related_manager_name="socialaccount_set",
+                child_model=SocialAccount,
+                parent_fk_name="user",
+            ),
+        }
+
+    def validate(self, attrs: dict) -> dict:
+        # social_accounts=[] 는 마지막 SA cascade 를 트리거해 같은 user 의 EA 전체를 삭제함.
+        # 같은 PATCH 의 email_addresses 입력은 cascade 로 즉시 사라져 의도와 다른 결과가 되므로,
+        # 실제로 cascade 가 발생하는 경우(기존 SA 존재 + SA=[] + EA 입력 있음)에만 거부.
+        if (
+            attrs.get("socialaccount_set") == []
+            and attrs.get("emailaddress_set")
+            and self.instance is not None
+            and self.instance.socialaccount_set.all()
+        ):
+            msg = "모든 SocialAccount 를 제거하면 EmailAddress 도 cascade 로 삭제됩니다 — 같은 PATCH 에서 EmailAddress 를 함께 변경할 수 없습니다."
+            raise serializers.ValidationError(msg)
+        return attrs
+
+    def validate_social_accounts(self, value: list[dict]) -> list[dict]:
+        # SocialAccount는 nested에서 delete-only — 모든 입력 id 가 이 유저의 기존 SA 와 매칭돼야 함.
+        # PATCH(partial=True) 에서는 nested 의 required 가 풀려 id 가 없을 수 있음 — 명시적으로 거부.
+        if any("id" not in item for item in value):
+            raise serializers.ValidationError("SocialAccount는 nested API에서 생성할 수 없습니다 (id 필수).")
+        provided_ids = {item["id"] for item in value}
+        # UserAdminViewSet 의 prefetch_related 캐시 활용.
+        existing_ids = {sa.id for sa in self.instance.socialaccount_set.all()} if self.instance else set()
+        if unknown := provided_ids - existing_ids:
+            msg = f"존재하지 않거나 이 유저의 것이 아닌 SocialAccount id: {sorted(map(str, unknown))}"
+            raise serializers.ValidationError(msg)
+        return value
 
     def create(self, validated_data: dict[str, typing.Any]) -> UserExt:
         password = generate_random_password()
         self._generated_password = password
-        return UserExt.objects.create_user(**validated_data, password=password)
+        nested_data = {k: validated_data.pop(k, []) or [] for k in self.Meta.nested_fields}
+        instance = UserExt.objects.create_user(**validated_data, password=password)
+        self._apply_nested_sync(instance, nested_data)
+        return instance
+
+    def _apply_nested_sync(self, instance: UserExt, nested_data: dict[str, list[dict] | None]) -> None:
+        # SocialAccount는 delete-only — 기존 set 에서 input 에 없는 것만 삭제.
+        sa_data = nested_data.pop("socialaccount_set", None)
+        super()._apply_nested_sync(instance, nested_data)
+
+        if sa_data is None:
+            return
+
+        provided_ids = {item["id"] for item in sa_data}
+        # prefetch_related 캐시 활용.
+        existing_ids = {sa.id for sa in instance.socialaccount_set.all()}
+        if to_delete_ids := existing_ids - provided_ids:
+            delete_social_accounts_and_cleanup_user_emails(SocialAccount.objects.filter(id__in=to_delete_ids))
 
 
 class UserAdminSignInSerializerData(typing.TypedDict):

app/admin_api/services/socialaccount.py

@@ -0,0 +1,14 @@
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialAccount
+from django.db import transaction
+from django.db.models import QuerySet
+
+
+def delete_social_accounts_and_cleanup_user_emails(social_accounts: QuerySet[SocialAccount]) -> None:
+    with transaction.atomic():
+        if not (affected_user_ids := set(social_accounts.values_list("user_id", flat=True))):
+            return
+        social_accounts.delete()
+        EmailAddress.objects.filter(user_id__in=affected_user_ids).exclude(
+            user_id__in=SocialAccount.objects.filter(user_id__in=affected_user_ids).values("user_id")
+        ).delete()