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 isinstance( file, __.io.TextIOBase ): 

83 content = file.read( ) 

84 configuration = __.tomli.loads( content ) 

85 else: 

86 configuration = await _io.acquire_text_file_async( 

87 file, deserializer = __.tomli.loads ) 

88 includes = await self._acquire_includes( 

89 application_name, 

90 directories, 

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

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

93 for edit in edits: edit( configuration ) 

94 return __.accret.Dictionary( configuration ) 

95 

96 async def _acquire_includes( 

97 self, 

98 application_name: str, 

99 directories: __.pdirs.PlatformDirs, 

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

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

102 locations = tuple( 

103 __.Path( spec.format( 

104 user_configuration = directories.user_config_path, 

105 user_home = __.Path.home( ), 

106 application_name = application_name ) ) 

107 for spec in specs ) 

108 iterables = tuple( 

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

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

111 for location in locations ) 

112 return await _io.acquire_text_files_async( 

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

114 deserializer = __.tomli.loads ) 

115 

116 def _discover_copy_template( 

117 self, 

118 directories: __.pdirs.PlatformDirs, 

119 distribution: _distribution.Information, 

120 ) -> __.Path: 

121 file = directories.user_config_path / self.main_filename 

122 if not file.exists( ): 

123 __.shutil.copyfile( 

124 distribution.provide_data_location( 

125 'configuration', self.main_filename ), file ) 

126 return file