Coverage for src / c41811 / config / processor / os_env.py: 100%

1# cython: language_level = 3 # noqa: ERA001 

2 

3 

4""" 

5环境变量配置数据处理器 

6 

7.. versionadded:: 0.2.0 

8""" 

9 

10import os 

11from collections import OrderedDict 

12from copy import deepcopy 

13from typing import Any 

14from typing import override 

15 

16from ..abc import ABCConfigFile 

17from ..abc import ABCSLProcessorPool 

18from ..basic.core import ConfigFile 

19from ..basic.environment import EnvironmentConfigData 

20from ..main import BasicConfigSL 

21 

22 

23class OSEnvSL(BasicConfigSL): 

24 """:py:data:`os.environ` 格式处理器""" 

25 

26 def __init__(self, *, reg_alias: str | None = None, prefix: str = "", strip_prefix: bool = False): 

27 """ 

28 :param reg_alias: sl处理器注册别名 

29 :type reg_alias: str | None 

30 :param prefix: (从环境变量)导出的环境变量前缀，留空则为所有 

31 :type prefix: str 

32 :param strip_prefix: (从环境变量)导出时是否去除前缀，导入(到环境变量)时会自动加回 

33 :type strip_prefix: bool 

34 

35 .. versionchanged:: 0.3.0 

36 添加参数 ``prefix`` 

37 添加参数 ``strip_prefix`` 

38 """ # noqa: RUF002, D205 

39 super().__init__(reg_alias=reg_alias) 

40 self.prefix = prefix 

41 self.strip_prefix = strip_prefix 

42 

43 @property 

44 @override 

45 def processor_reg_name(self) -> str: 

46 return "os.environ" 

47 

48 @property 

49 @override 

50 def supported_file_patterns(self) -> tuple[str, ...]: 

51 return ".os.env", ".os.environ" 

52 

53 supported_file_classes = [ConfigFile] # noqa: RUF012 

54 

55 @override 

56 def save( 

57 self, 

58 processor_pool: ABCSLProcessorPool, 

59 config_file: ABCConfigFile[EnvironmentConfigData], 

60 root_path: str, 

61 namespace: str, 

62 file_name: str, 

63 *args: Any, 

64 **kwargs: Any, 

65 ) -> None: 

66 cfg: EnvironmentConfigData = config_file.config 

67 diff = cfg.difference 

68 is_striped = self.strip_prefix and self.prefix 

69 

70 for updated in deepcopy(diff.updated): 

71 env_key = f"{self.prefix}{updated}" if is_striped else updated 

72 os.environ[env_key] = cfg[updated] 

73 diff.updated.add(updated) 

74 for removed in deepcopy(diff.removed): 

75 env_key = f"{self.prefix}{removed}" if is_striped else removed 

76 del os.environ[env_key] 

77 diff.removed.remove(removed) 

78 

79 @override 

80 def load( 

81 self, 

82 processor_pool: ABCSLProcessorPool, 

83 root_path: str, 

84 namespace: str, 

85 file_name: str, 

86 *args: Any, 

87 **kwargs: Any, 

88 ) -> ConfigFile[EnvironmentConfigData]: 

89 if not self.prefix: 

90 return ConfigFile(EnvironmentConfigData(OrderedDict(os.environ))) 

91 filtered_env = OrderedDict() 

92 for key, value in os.environ.items(): 

93 if key.startswith(self.prefix): 

94 config_key = key[len(self.prefix) :] if self.strip_prefix else key 

95 filtered_env[config_key] = value 

96 

97 return ConfigFile(EnvironmentConfigData(filtered_env)) 

98 

99 

100__all__ = ("OSEnvSL",)