refactor(open api v2 - list user): reduce unnecessary fields in SQL S… (#2312)

nannan00 · web-flow · commit c99bec043ba5 · 2025-12-25T11:40:01.000+08:00
diff --git a/src/bk-user/bkuser/apis/open_v2/urls.py b/src/bk-user/bkuser/apis/open_v2/urls.py
@@ -38,7 +38,7 @@
         views.ProfileDepartmentListApi.as_view(),
         name="open_v2.list_profile_departments",
     ),
-    # 与上面 API 一样，只是兼容了缺少末尾 / 的情况 （ESB yaml 配置里是该情况）
+    # 与上面 API 一样，只是兼容了缺少末尾 / 的情况（ESB yaml 配置里是该情况）
     path(
         "profiles/<str:lookup_value>/departments",
         views.ProfileDepartmentListApi.as_view(),
@@ -66,7 +66,7 @@
         views.DepartmentProfileListApi.as_view(),
         name="open_v2.list_department_profiles",
     ),
-    # 与上面 API 一样，只是兼容了缺少末尾 / 的情况 （ESB yaml 配置里是该情况）
+    # 与上面 API 一样，只是兼容了缺少末尾 / 的情况（ESB yaml 配置里是该情况）
     path(
         "departments/<str:id>/profiles",
         views.DepartmentProfileListApi.as_view(),
diff --git a/src/bk-user/bkuser/apis/open_v2/views/profilers.py b/src/bk-user/bkuser/apis/open_v2/views/profilers.py
@@ -86,80 +86,202 @@ def build_user_infos(self, tenant_users: QuerySet[TenantUser], fields: List[str]
                              且必须保证 select_related("data_source_user")
         :param fields: 对外的用户字段列表，空时表示所有用户字段都对外
         """
-        # 按需提前获取用户 Leader 信息 和 用户部门信息
+        # 标准化字段列表
+        fields = self._get_default_fields(fields)
+
+        # 优化查询：只查询需要的数据库字段
+        tenant_users = self._optimize_queryset(tenant_users, fields)
+
+        # 批量预加载关联数据
+        context = self._prepare_context(tenant_users, fields)
+
+        # 构建每个用户的信息
+        return [self._build_single_user_info(tenant_user, fields, context) for tenant_user in tenant_users]
+
+    @staticmethod
+    def _get_default_fields(fields: List[str]) -> List[str]:
+        """获取默认字段列表"""
+        return fields or [
+            "id",
+            "username",
+            "display_name",
+            "email",
+            "telephone",
+            "country_code",
+            "time_zone",
+            "language",
+            "wx_userid",
+            "domain",
+            "category_id",
+            "status",
+            "extras",
+            "position",
+            "leader",
+            "departments",
+        ]
+
+    @staticmethod
+    def _get_db_field_map() -> Dict[str, List[str]]:
+        """获取字段到数据库字段的映射"""
+        return {
+            "id": ["data_source_user__id"],
+            "username": ["id"],
+            "display_name": ["data_source_user__full_name"],
+            "email": ["custom_email", "is_inherited_email", "data_source_user__email"],
+            "telephone": ["custom_phone", "is_inherited_phone", "data_source_user__phone"],
+            "country_code": [
+                "custom_phone_country_code",
+                "is_inherited_phone",
+                "data_source_user__phone_country_code",
+            ],
+            "time_zone": ["time_zone"],
+            "language": ["language"],
+            "wx_userid": ["wx_userid"],
+            "domain": ["data_source_id", "tenant_id"],
+            "category_id": ["data_source_id"],
+            "status": ["status"],
+            "extras": ["data_source_user__extras"],
+            "position": ["data_source_user__extras"],
+            "leader": ["data_source_user__id"],
+            "departments": ["data_source_user__id"],
+        }
+
+    def _optimize_queryset(self, tenant_users: QuerySet[TenantUser], fields: List[str]) -> QuerySet[TenantUser]:
+        """优化查询，只查询需要的数据库字段"""
+        db_field_map = self._get_db_field_map()
+        only_fields = {"id"}
+        for f in fields:
+            only_fields.update(db_field_map.get(f, []))
+        return tenant_users.only(*only_fields)
+
+    def _prepare_context(self, tenant_users: QuerySet[TenantUser], fields: List[str]) -> Dict[str, Any]:
+        """预加载关联数据，减少数据库查询"""
         data_source_user_ids = [i.data_source_user.id for i in tenant_users]
-        leader_map = self._get_leader_map(data_source_user_ids) if not fields or "leader" in fields else {}
-        department_map = (
-            self._get_department_map(data_source_user_ids) if not fields or "departments" in fields else {}
-        )
 
-        collaboration_field_mapping = self.get_collaboration_field_mapping()
-
-        user_infos = []
-        for tenant_user in tenant_users:
-            # 手机号和手机区号
-            phone, phone_country_code = tenant_user.phone_info
-
-            # 自定义字段
-            source_extras = tenant_user.data_source_user.extras
-            # 协同时按照协同租户配置的用户自定义字段进行输出
-            ds_owner_tenant_id = tenant_user.data_source.owner_tenant_id
-            if ds_owner_tenant_id != tenant_user.tenant_id:
-                extras = {
-                    collaboration_field_mapping[(ds_owner_tenant_id, k)]: v
-                    for k, v in source_extras.items()
-                    if (ds_owner_tenant_id, k) in collaboration_field_mapping
-                }
-            else:
-                extras = source_extras
-
-            # 不会放大查询的字段
-            user_info = {
-                "id": tenant_user.data_source_user.id,
-                # 租户用户 ID 即为对外的 username / bk_username
-                "username": tenant_user.id,
-                "display_name": tenant_user.data_source_user.full_name,
-                "email": tenant_user.email,
-                "telephone": phone,
-                "country_code": phone_country_code,
-                "iso_code": _phone_country_code_to_iso_code(phone_country_code),
-                "time_zone": tenant_user.time_zone,
-                "language": tenant_user.language,
-                "wx_userid": tenant_user.wx_userid,
-                "domain": self.get_domain(tenant_user.data_source_id, tenant_user.tenant_id),
-                "category_id": tenant_user.data_source_id,
-                "status": TENANT_USER_STATUS_TO_PROFILE_STATUS_MAP.get(tenant_user.status, tenant_user.status),
-                "enabled": True,
-                "extras": extras,
-                # 旧版本内置字段，新版本迁移在自定义字段里
-                "position": int(source_extras.get("position", 0)),
-                # 总是返回固定值
-                "logo": "",
-                "staff_status": "IN",
-            }
+        return {
+            "leader_map": self._get_leader_map(data_source_user_ids) if "leader" in fields else {},
+            "department_map": self._get_department_map(data_source_user_ids) if "departments" in fields else {},
+            "collaboration_field_mapping": (
+                self.get_collaboration_field_mapping() if "extras" in fields or "position" in fields else {}
+            ),
+        }
 
-            # 指定对外字段，则只返回指定的字段
-            if fields:
-                user_info = {k: v for k, v in user_info.items() if k in fields}
-                # 由于 leader 需要额外计算，因此特殊分支处理
-                if "leader" in fields:
-                    user_info["leader"] = leader_map.get((tenant_user.tenant.id, tenant_user.data_source_user.id))
-                # 由于 department 需要额外计算，因此特殊分支处理
-                if "departments" in fields:
-                    user_info["departments"] = department_map.get(
-                        (tenant_user.tenant.id, tenant_user.data_source_user.id)
-                    )
+    def _build_single_user_info(
+        self, tenant_user: TenantUser, fields: List[str], context: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """构建单个用户的信息"""
+        user_info: Dict[str, Any] = {}
 
-                user_infos.append(user_info)
-                continue
+        # 添加基础字段
+        self._add_basic_fields(user_info, tenant_user, fields)
 
-            # 未指定字段，则关联字段也要返回
-            user_info["leader"] = leader_map.get((tenant_user.tenant.id, tenant_user.data_source_user.id))
-            user_info["departments"] = department_map.get((tenant_user.tenant.id, tenant_user.data_source_user.id))
+        # 添加电话相关字段
+        self._add_phone_fields(user_info, tenant_user, fields)
 
-            user_infos.append(user_info)
+        # 添加简单字段
+        self._add_simple_fields(user_info, tenant_user, fields)
+
+        # 添加自定义字段
+        self._add_extras_fields(user_info, tenant_user, fields, context["collaboration_field_mapping"])
+
+        # 添加固定值字段
+        user_info["logo"] = ""
+        user_info["staff_status"] = "IN"
+
+        # 添加关联字段
+        self._add_relation_fields(user_info, tenant_user, fields, context)
+
+        return user_info
 
-        return user_infos
+    @staticmethod
+    def _add_basic_fields(user_info: Dict[str, Any], tenant_user: TenantUser, fields: List[str]) -> None:
+        """添加基础字段"""
+        if "id" in fields:
+            user_info["id"] = tenant_user.data_source_user.id
+        if "username" in fields:
+            user_info["username"] = tenant_user.id
+        if "display_name" in fields:
+            user_info["display_name"] = tenant_user.data_source_user.full_name
+        if "email" in fields:
+            user_info["email"] = tenant_user.email
+
+    @staticmethod
+    def _add_phone_fields(user_info: Dict[str, Any], tenant_user: TenantUser, fields: List[str]) -> None:
+        """添加电话相关字段"""
+        if not any(f in fields for f in ["telephone", "country_code", "iso_code"]):
+            return
+
+        phone, phone_country_code = tenant_user.phone_info
+        if "telephone" in fields:
+            user_info["telephone"] = phone
+        if "country_code" in fields:
+            user_info["country_code"] = phone_country_code
+        if "iso_code" in fields:
+            user_info["iso_code"] = _phone_country_code_to_iso_code(phone_country_code)
+
+    def _add_simple_fields(self, user_info: Dict[str, Any], tenant_user: TenantUser, fields: List[str]) -> None:
+        """添加简单字段"""
+        simple_field_mapping = {
+            "time_zone": lambda: tenant_user.time_zone,
+            "language": lambda: tenant_user.language,
+            "wx_userid": lambda: tenant_user.wx_userid,
+            "domain": lambda: self.get_domain(tenant_user.data_source_id, tenant_user.tenant_id),
+            "category_id": lambda: tenant_user.data_source_id,
+            "status": lambda: TENANT_USER_STATUS_TO_PROFILE_STATUS_MAP.get(tenant_user.status, tenant_user.status),
+        }
+
+        for field_name, value_func in simple_field_mapping.items():
+            if field_name in fields:
+                user_info[field_name] = value_func()
+
+    @staticmethod
+    def _add_extras_fields(
+        user_info: Dict[str, Any],
+        tenant_user: TenantUser,
+        fields: List[str],
+        collaboration_field_mapping: Dict[Tuple[str, str], str],
+    ) -> None:
+        """添加自定义字段"""
+        if not ("extras" in fields or "position" in fields):
+            return
+
+        source_extras = tenant_user.data_source_user.extras
+        extras = TenantUserListToUserInfosMixin._get_collaboration_extras(
+            source_extras, tenant_user, collaboration_field_mapping
+        )
+
+        if "extras" in fields:
+            user_info["extras"] = extras
+        if "position" in fields:
+            user_info["position"] = int(source_extras.get("position", 0))
+
+    @staticmethod
+    def _get_collaboration_extras(
+        source_extras: Dict[str, Any],
+        tenant_user: TenantUser,
+        collaboration_field_mapping: Dict[Tuple[str, str], str],
+    ) -> Dict[str, Any]:
+        """获取协同场景下的自定义字段"""
+        ds_owner_tenant_id = tenant_user.data_source.owner_tenant_id
+        if ds_owner_tenant_id != tenant_user.tenant_id:
+            return {
+                collaboration_field_mapping[(ds_owner_tenant_id, k)]: v
+                for k, v in source_extras.items()
+                if (ds_owner_tenant_id, k) in collaboration_field_mapping
+            }
+        return source_extras
+
+    @staticmethod
+    def _add_relation_fields(
+        user_info: Dict[str, Any], tenant_user: TenantUser, fields: List[str], context: Dict[str, Any]
+    ) -> None:
+        """添加关联字段"""
+        if "leader" in fields:
+            user_info["leader"] = context["leader_map"].get((tenant_user.tenant.id, tenant_user.data_source_user.id))
+        if "departments" in fields:
+            user_info["departments"] = context["department_map"].get(
+                (tenant_user.tenant.id, tenant_user.data_source_user.id)
+            )
 
     @staticmethod
     def _get_leader_map(data_source_user_ids: List[int]) -> Dict[Tuple[str, int], List[Dict[str, Any]]]:
@@ -480,7 +602,7 @@ def get(self, request, *args, **kwargs):
         slz.is_valid(raise_exception=True)
         params = slz.validated_data
 
-        lookup_filter = {}
+        lookup_filter: Dict[str, Any] = {}
         if params["lookup_field"] == "username":
             # username 其实就是新的租户用户 ID，形式如 admin / admin@qq.com / uuid4 / nanoid
             lookup_filter["id"] = kwargs["lookup_value"]
diff --git a/src/bk-user/bkuser/common/cache.py b/src/bk-user/bkuser/common/cache.py
@@ -18,7 +18,6 @@
 import functools
 
 from blue_krill.data_types.enum import EnumField, StrStructuredEnum
-from django.core.cache import cache as default_cache
 from django.core.cache import caches
 from django.core.cache.backends.base import DEFAULT_TIMEOUT
 
@@ -73,16 +72,16 @@ def _method_key_function(_, *args, **kwargs):
 # cached 和 cachedmethod 其 key 的生成方法可以满足大部分情况下不冲突，但有以下几种情况可能会冲突
 # (1) 对于类的实例方法，由于缓存 key 只用到方法的自定义参数，
 #     若 key 的区分需要用到 self.{attr}，则需要重新自定义，否则相同方法参数时会冲突
-# (2) 虽然模块名+方法名作为了 key 的前缀，但由于是字符串拼接，
+# (2) 虽然模块名 + 方法名作为了 key 的前缀，但由于是字符串拼接，
 #     有极少概率会出现拼接出来的结果一样的情况而导致冲突
 # (3) key 的字符串拼接，若参数里的值包含分隔符 "|"，有可能出现冲突
 # (4) 生成 key 时做了字符串转换，对于某些对象可能 str() 后相同，
 #     建议参数类型为：str/bool/int/tuple/List[base_type]/Dict[base_type]
-def cached(cache=default_cache, key_function=_default_key_function, timeout=DEFAULT_TIMEOUT):
+def cached(cache_name=CacheEnum.DEFAULT.value, key_function=_default_key_function, timeout=DEFAULT_TIMEOUT):
     """Decorator to wrap a function with a memorizing callable that saves results in a cache.
     cache param usage:
         from django.core.cache import caches
-        @cached(cache=caches[CacheEnum.REDIS], ...)
+        @cached(cache_name=CacheEnum.REDIS, ...)
     """
 
     def decorator(func):
@@ -92,14 +91,14 @@ def wrapper(*args, **kwargs):
             namespace = f"{func.__module__}:{func.__name__}"
             key = f"{CacheKeyPrefixEnum.AUTO}:{namespace}:{custom_key}"
 
-            return cache.get_or_set(key, lambda: func(*args, **kwargs), timeout)
+            return caches[str(cache_name)].get_or_set(key, lambda: func(*args, **kwargs), timeout)
 
         return wrapper
 
     return decorator
 
 
-def cachedmethod(cache=default_cache, key_function=_method_key_function, timeout=DEFAULT_TIMEOUT):
+def cachedmethod(cache_name=CacheEnum.DEFAULT.value, key_function=_method_key_function, timeout=DEFAULT_TIMEOUT):
     """Decorator to wrap a class or instance method with a memorizing
     callable that saves results in a cache.
     """
@@ -111,7 +110,7 @@ def wrapper(self, *args, **kwargs):
             namespace = f"{method.__module__}:{method.__qualname__}"
             key = f"{CacheKeyPrefixEnum.AUTO}:{namespace}:{custom_key}"
 
-            return cache.get_or_set(key, lambda: method(self, *args, **kwargs), timeout)
+            return caches[str(cache_name)].get_or_set(key, lambda: method(self, *args, **kwargs), timeout)
 
         return wrapper
 
@@ -122,7 +121,7 @@ class Cache:
     """
     Cache 用于避免直接使用 Django Caches 时导致不同场景的前缀 Key 冲突问题，
     使用各个场景更专注于自身业务逻辑缓存和 key 生成，Cache 所有方法都基于
-    Django Cache 的 BaseCache ，只封装了项目所需方法
+    Django Cache 的 BaseCache，只封装了项目所需方法
     """
 
     def __init__(self, type_, key_prefix):
diff --git a/src/bk-user/tests/apis/open_v2/conftest.py b/src/bk-user/tests/apis/open_v2/conftest.py
@@ -21,7 +21,7 @@
 from bkuser.apps.tenant.models import CollaborationStrategy
 from bkuser.plugins.general.models import GeneralDataSourcePluginConfig
 from bkuser.plugins.local.models import LocalDataSourcePluginConfig
-from django.core.cache import cache
+from django.core.cache import caches
 
 from tests.test_utils.data_source import init_data_source_users_depts_and_relations
 from tests.test_utils.helpers import generate_random_string
@@ -33,9 +33,12 @@
 @pytest.fixture(autouse=True)
 def _clear_cache():
     """在每个测试前清除缓存，避免缓存的方法返回值影响测试结果"""
-    cache.clear()
+    # 清除所有配置的缓存
+    for cache_name in caches:
+        caches[cache_name].clear()
     yield
-    cache.clear()
+    for cache_name in caches:
+        caches[cache_name].clear()
 
 
 @pytest.fixture
