Blame view

config/tenant_config_loader.py 2.63 KB
0064e946   tangwang   feat: 增量索引服务、租户配置...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
  """
  租户配置加载器。
  
  从统一配置文件(config.yaml)加载租户配置,包括主语言和翻译配置。
  """
  
  import logging
  from typing import Dict, Any, Optional
  
  logger = logging.getLogger(__name__)
  
  
  class TenantConfigLoader:
      """租户配置加载器。"""
  
      def __init__(self):
          """初始化租户配置加载器。"""
          self._config: Optional[Dict[str, Any]] = None
  
      def load_config(self) -> Dict[str, Any]:
          """
          加载租户配置(从统一配置文件)。
  
          Returns:
              租户配置字典,格式:{"tenants": {...}, "default": {...}}
          """
          if self._config is not None:
              return self._config
  
          try:
              from config import ConfigLoader
              config_loader = ConfigLoader()
              search_config = config_loader.load_config()
              self._config = search_config.tenant_config
              logger.info("Loaded tenant config from unified config.yaml")
              return self._config
          except Exception as e:
              logger.error(f"Failed to load tenant config: {e}", exc_info=True)
              # 返回默认配置
              self._config = {
                  "default": {
                      "primary_language": "zh",
                      "translate_to_en": True,
                      "translate_to_zh": False
                  },
                  "tenants": {}
              }
              return self._config
  
      def get_tenant_config(self, tenant_id: str) -> Dict[str, Any]:
          """
          获取指定租户的配置。
  
          Args:
              tenant_id: 租户ID
  
          Returns:
              租户配置字典,如果租户不存在则返回默认配置
          """
          config = self.load_config()
          tenant_id_str = str(tenant_id)
          
          tenants = config.get("tenants", {})
          if tenant_id_str in tenants:
              return tenants[tenant_id_str]
          else:
              logger.debug(f"Tenant {tenant_id} not found in config, using default")
              return config.get("default", {
                  "primary_language": "zh",
                  "translate_to_en": True,
                  "translate_to_zh": False
              })
  
      def reload(self):
          """重新加载配置(用于配置更新)。"""
          self._config = None
          return self.load_config()
  
  
  # 全局实例
  _tenant_config_loader: Optional[TenantConfigLoader] = None
  
  
  def get_tenant_config_loader() -> TenantConfigLoader:
      """获取全局租户配置加载器实例。"""
      global _tenant_config_loader
      if _tenant_config_loader is None:
          _tenant_config_loader = TenantConfigLoader()
      return _tenant_config_loader