Coverage for sources/appcore/configuration.py: 100%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Fundamental configuration. ''' 

22# TODO: Add configuration validation (schema validation, type checking) 

23 

24 

25from . import __ 

26from . import dictedits as _dictedits 

27from . import distribution as _distribution 

28from . import exceptions as _exceptions 

29from . import io as _io 

30 

31 

32class EnablementTristate( __.enum.Enum ): # TODO: Python 3.11: StrEnum 

33 ''' Disable, enable, or retain the natural state? ''' 

34 

35 Disable = 'disable' 

36 Retain = 'retain' 

37 Enable = 'enable' 

38 

39 def __bool__( self ) -> bool: 

40 if self.Disable is self: return False 

41 if self.Enable is self: return True 

42 raise _exceptions.OperationInvalidity( # noqa: TRY003 

43 'inert enablement tristate', 'boolean translation' ) 

44 

45 def is_retain( self ) -> bool: 

46 ''' Does enum indicate a retain state? ''' 

47 return self.Retain is self 

48 

49 

50class AcquirerAbc( __.immut.DataclassProtocol, __.typx.Protocol ): 

51 ''' Abstract base class for configuration acquirers. ''' 

52 

53 @__.abc.abstractmethod 

54 async def __call__( 

55 self, 

56 application_name: str, 

57 directories: __.pdirs.PlatformDirs, 

58 distribution: _distribution.Information, 

59 edits: _dictedits.Edits = ( ), 

60 file: __.Absential[ __.Path | __.io.TextIOBase ] = __.absent, 

61 ) -> __.accret.Dictionary[ str, __.typx.Any ]: 

62 ''' Provides configuration as accretive dictionary. ''' 

63 raise NotImplementedError # pragma: no cover 

64 

65 

66class TomlAcquirer( AcquirerAbc ): 

67 ''' Acquires configuration data from TOML data files. ''' 

68 

69 main_filename: str = 'general.toml' 

70 includes_name: str = 'includes' 

71 

72 async def __call__( 

73 self, 

74 application_name: str, 

75 directories: __.pdirs.PlatformDirs, 

76 distribution: _distribution.Information, 

77 edits: _dictedits.Edits = ( ), 

78 file: __.Absential[ __.Path | __.io.TextIOBase ] = __.absent, 

79 ) -> __.accret.Dictionary[ str, __.typx.Any ]: 

80 if __.is_absent( file ): 

81 file = self._discover_copy_template( directories, distribution ) 

82 if __.is_absent( file ): return __.accret.Dictionary( { } ) 

83 if isinstance( file, __.io.TextIOBase ): 

84 content = file.read( ) 

85 configuration = __.tomli.loads( content ) 

86 else: 

87 configuration = await _io.acquire_text_file_async( 

88 file, deserializer = __.tomli.loads ) 

89 includes = await self._acquire_includes( 

90 application_name, 

91 directories, 

92 configuration.get( self.includes_name, ( ) ) ) 

93 for include in includes: configuration.update( include ) 

94 for edit in edits: edit( configuration ) 

95 return __.accret.Dictionary( configuration ) 

96 

97 async def _acquire_includes( 

98 self, 

99 application_name: str, 

100 directories: __.pdirs.PlatformDirs, 

101 specs: tuple[ str, ... ], 

102 ) -> __.cabc.Sequence[ dict[ str, __.typx.Any ] ]: 

103 locations = tuple( 

104 __.Path( spec.format( 

105 user_configuration = directories.user_config_path, 

106 user_home = __.Path.home( ), 

107 application_name = application_name ) ) 

108 for spec in specs ) 

109 iterables = tuple( 

110 ( location.glob( '*.toml' ) 

111 if location.is_dir( ) else ( location, ) ) 

112 for location in locations ) 

113 return await _io.acquire_text_files_async( 

114 *( file for file in __.itert.chain.from_iterable( iterables ) ), 

115 deserializer = __.tomli.loads ) 

116 

117 def _discover_copy_template( 

118 self, 

119 directories: __.pdirs.PlatformDirs, 

120 distribution: _distribution.Information, 

121 ) -> __.Absential[ __.Path ]: 

122 file = directories.user_config_path / self.main_filename 

123 if not file.exists( ): 

124 template_location = distribution.provide_data_location( 

125 'configuration', self.main_filename ) 

126 if template_location.exists( ): 

127 __.shutil.copyfile( template_location, file ) 

128 else: return __.absent 

129 return file