Coverage for sources/mimeogram/__/configuration.py: 100%

40 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-02-17 22:34 +0000

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 

23 

24from . import imports as __ 

25from . import dictedits as _dictedits 

26from . import distribution as _distribution 

27from . import exceptions as _exceptions 

28from . import io as _io 

29 

30 

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

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

33 

34 Disable = 'disable' 

35 Retain = 'retain' 

36 Enable = 'enable' 

37 

38 def __bool__( self ) -> bool: 

39 if self.Disable is self: return False 

40 if self.Enable is self: return True 

41 raise _exceptions.OperationInvalidity( 

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

43 

44 def is_retain( self ) -> bool: 

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

46 return self.Retain is self 

47 

48 

49async def acquire( 

50 application_name: str, 

51 directories: __.PlatformDirs, 

52 distribution: _distribution.Information, 

53 edits: _dictedits.Edits = ( ), 

54 file: __.Absential[ __.Path ] = __.absent, 

55) -> __.AccretiveDictionary[ str, __.typx.Any ]: 

56 ''' Loads configuration as dictionary. ''' 

57 if __.is_absent( file ): 

58 file = _discover_copy_template( directories, distribution ) 

59 configuration = await _acquire( file ) 

60 includes = await _acquire_includes( 

61 application_name, directories, configuration.get( 'includes', ( ) ) ) 

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

63 for edit in edits: edit( configuration ) 

64 return __.AccretiveDictionary( configuration ) 

65 

66 

67async def _acquire( file: __.Path ) -> __.NominativeDictionary: 

68 from aiofiles import open as open_ 

69 from tomli import loads 

70 async with open_( file ) as stream: 

71 return loads( await stream.read( ) ) 

72 

73 

74async def _acquire_includes( 

75 application_name: str, 

76 directories: __.PlatformDirs, 

77 specs: tuple[ str, ... ] 

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

79 from itertools import chain 

80 from tomli import loads 

81 locations = tuple( 

82 __.Path( spec.format( 

83 user_configuration = directories.user_config_path, 

84 user_home = __.Path.home( ), 

85 application_name = application_name ) ) 

86 for spec in specs ) 

87 iterables = tuple( 

88 ( location.glob( '*.toml' ) if location.is_dir( ) else ( location, ) ) 

89 for location in locations ) 

90 return await _io.acquire_text_files_async( 

91 *( file for file in chain.from_iterable( iterables ) ), 

92 deserializer = loads ) 

93 

94 

95def _discover_copy_template( 

96 directories: __.PlatformDirs, distribution: _distribution.Information 

97) -> __.Path: 

98 from shutil import copyfile 

99 file = directories.user_config_path / 'general.toml' 

100 if not file.exists( ): 

101 copyfile( 

102 distribution.provide_data_location( 

103 'configuration', 'general.toml' ), file ) 

104 return file