Coverage for configlayer/_file.py: 99%
69 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-08 11:36 +0000
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-08 11:36 +0000
1"""Internal config layer file support structure"""
2from pathlib import Path
3from configparser import ConfigParser
5from .utils import Locker
6from .types import path_t, mb_holder_t
7from .exceptions import InitError, FileError
10class File(Locker):
11 """File optional structure
12 Used in config support structure if file path provided, for local storage of configs"""
13 __slots__ = ('_cfg', 'path')
14 _used_paths: dict = dict() # Common fixed class variable (dict methods only)
15 path: Path
17 def __init__(self, cfg, path: path_t):
18 self._cfg = cfg
19 if (path := Path(path)) in self._used_paths:
20 raise InitError(f'Path "{path}" is already used in {self._used_paths[path]!r} config')
22 self.path = path
23 if path.exists():
24 self.load()
25 else:
26 self.save() # Create "empty" file for path correctness check, folder is not created
27 path.unlink() # Remove "empty" file
28 self._used_paths[path] = cfg.name
30 # Locks structure for changes with disabling attribute deletion
31 super().__init__(del_attr=False)
33 def __del__(self):
34 """Remove path from used at config deletion"""
35 if hasattr(self, 'path'):
36 self._used_paths.pop(self.path, None)
38 def __repr__(self):
39 return f'{self._cfg!r}.file'
41 def __str__(self):
42 return f'{self._cfg.name!r} {self._cfg.type_name} file support structure'
44 @staticmethod
45 def _exc(op):
46 def init_wrapper(func):
47 def method_wrapper(self, *args, **kwargs):
48 try:
49 return func(self, *args, **kwargs)
50 except FileNotFoundError as e:
51 msg = e.args[1]
52 except FileError as e:
53 msg = e
54 except Exception as e:
55 raise FileError(f'{op} "{self.path}" failed. {e!r}') from e
56 raise FileError(f'{op} "{self.path}" failed. {msg}')
57 return method_wrapper
58 return init_wrapper
60 def _get_config(self):
61 config = ConfigParser(default_section=f'_{self._cfg.def_sect}') # hidden reserved name
62 config.optionxform = str
63 return config
65 @_exc('Save to')
66 def save(self, sections: mb_holder_t[str] | None = None, *,
67 strict_defaults=False, strict_data=False):
68 """Save config to file (or only selected sections, excepting internal)
69 :arg sections: Selected section name(s) or all (if not provided)
70 :arg strict_defaults: Save all fields from default section (not skip equal to factory)
71 :arg strict_data: Save all fields from data sections (not skip equal to default)
72 :raise InputError: If wrong arguments provided
73 :raise IOExportError: If errors during export_config"""
74 config = self._get_config()
75 config.read_dict(self._cfg.io.export_config(sections, strict_defaults=strict_defaults,
76 strict_data=strict_data, typecast=True))
77 with self.path.open('w', encoding='utf-8') as file:
78 config.write(file)
80 @_exc('Load from')
81 def load(self, sections: mb_holder_t[str] | None = None):
82 """Load config from file (or only selected sections, excepting internal)
83 :arg sections: Selected section name(s) or all (if not provided)
84 :raise InputError: If wrong arguments provided
85 :raise IOImportError: If errors during import_config
86 :raise FileError: If file contains forbidden default section"""
87 config = self._get_config()
88 hidden_default_sect = f'_{self._cfg.def_sect}'
90 # Raw check for provided real default section (not used due to internal section impact)
91 error = False
92 with self.path.open('r', encoding='utf-8') as file:
93 config.read_file(file)
94 file.seek(0)
95 for line in file.readlines():
96 if line.startswith(f'[{hidden_default_sect}]'):
97 error = True
98 break
100 # Process data as dicts and raise an error if real default section provided
101 data = {k: dict(v) for k, v in config.items()}
102 if (wrong := data.pop(hidden_default_sect)) or error:
103 details = f'. Data: {wrong}' if wrong else ''
104 raise FileError(f'{hidden_default_sect!r} section is forbidden, but provided{details}')
105 self._cfg.io.import_config(data, sections)