feat: list data source leaders and departments #1188

src/bk-user/bkuser/apis/web/data_source/serializers.py

@@ -87,3 +87,21 @@ def validate_leader_ids(self, leader_ids):
 
 class UserCreateOutputSLZ(serializers.Serializer):
     id = serializers.CharField(help_text="数据源用户ID")
+
+
+class LeaderSearchInputSLZ(serializers.Serializer):
+    keyword = serializers.CharField(help_text="搜索关键字", required=False)
+
+
+class LeaderSearchOutputSLZ(serializers.Serializer):
+    id = serializers.CharField(help_text="上级ID")
+    username = serializers.CharField(help_text="上级名称")
+
+
+class DepartmentSearchInputSLZ(serializers.Serializer):
+    name = serializers.CharField(required=False, help_text="部门名称", allow_blank=True)
+
+
+class DepartmentSearchOutputSLZ(serializers.Serializer):
+    id = serializers.CharField(help_text="部门ID")
+    name = serializers.CharField(help_text="部门名称")

src/bk-user/bkuser/apis/web/data_source/urls.py

@@ -14,4 +14,6 @@
 
 urlpatterns = [
     path("<int:id>/users/", views.DataSourceUserListCreateApi.as_view(), name="data_source_user.list_create"),
+    path("<int:id>/leaders/", views.DataSourceLeadersListApi.as_view(), name="data_source_leaders.list"),
+    path("<int:id>/departments/", views.DataSourceDepartmentsListApi.as_view(), name="data_source_departments.list"),
 ]

src/bk-user/bkuser/apis/web/data_source/views.py

@@ -8,17 +8,22 @@
 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 specific language governing permissions and limitations under the License.
 """
+from django.db.models import Q
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import generics, status
 from rest_framework.response import Response
 
 from bkuser.apis.web.data_source.serializers import (
+    DepartmentSearchInputSLZ,
+    DepartmentSearchOutputSLZ,
+    LeaderSearchInputSLZ,
+    LeaderSearchOutputSLZ,
     UserCreateInputSLZ,
     UserCreateOutputSLZ,
     UserSearchInputSLZ,
     UserSearchOutputSLZ,
 )
-from bkuser.apps.data_source.models import DataSource, DataSourceUser
+from bkuser.apps.data_source.models import DataSource, DataSourceDepartment, DataSourceUser
 from bkuser.biz.data_source_organization import (
     DataSourceOrganizationHandler,
     DataSourceUserBaseInfo,
@@ -99,3 +104,60 @@ def post(self, request, *args, **kwargs):
             data_source=data_source, base_user_info=base_user_info, relation_info=relation_info
         )
         return Response(UserCreateOutputSLZ(instance={"id": user_id}).data)
+
+
+class DataSourceLeadersListApi(generics.ListAPIView):
+    serializer_class = LeaderSearchOutputSLZ
+
+    def get_queryset(self):
+        slz = LeaderSearchInputSLZ(data=self.request.query_params)
+        slz.is_valid(raise_exception=True)
+        data = slz.validated_data
+
+        # 校验数据源是否存在
+        data_source = DataSource.objects.filter(id=self.kwargs["id"]).first()
+        if not data_source:
+            raise error_codes.DATA_SOURCE_NOT_EXIST
+
+        queryset = DataSourceUser.objects.filter(data_source=data_source)
+        if keyword := data.get("keyword"):
+            queryset = queryset.filter(Q(username__icontains=keyword) | Q(full_name__icontains=keyword))
+
+        return queryset
+
+    @swagger_auto_schema(
+        operation_description="数据源上级列表",
+        query_serializer=LeaderSearchInputSLZ(),
+        responses={status.HTTP_200_OK: LeaderSearchOutputSLZ(many=True)},
+    )
+    def get(self, request, *args, **kwargs):
+        return self.list(request, *args, **kwargs)
+
+
+class DataSourceDepartmentsListApi(generics.ListAPIView):
+    serializer_class = DepartmentSearchOutputSLZ
+
+    def get_queryset(self):
+        slz = DepartmentSearchInputSLZ(data=self.request.query_params)
+        slz.is_valid(raise_exception=True)
+        data = slz.validated_data
+
+        # 校验数据源是否存在
+        data_source = DataSource.objects.filter(id=self.kwargs["id"]).first()
+        if not data_source:
+            raise error_codes.DATA_SOURCE_NOT_EXIST
+
+        queryset = DataSourceDepartment.objects.filter(data_source=data_source)
+
+        if name := data.get("name"):
+            queryset = queryset.filter(name__icontains=name)
+
+        return queryset
+
+    @swagger_auto_schema(
+        operation_description="数据源部门列表",
+        query_serializer=DepartmentSearchInputSLZ(),
+        responses={status.HTTP_200_OK: DepartmentSearchOutputSLZ(many=True)},
+    )
+    def get(self, request, *args, **kwargs):
+        return self.list(request, *args, **kwargs)