Coverage for configzen/sources.py: 39%
153 statements
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-15 02:42 +0000
1"""Sources and destinations that hold the configuration data."""
3from __future__ import annotations 1abcdefgh
5from abc import ABCMeta, abstractmethod 1abcdefgh
6from functools import singledispatch 1abcdefgh
7from io import BytesIO, StringIO 1abcdefgh
8from os import PathLike 1abcdefgh
9from pathlib import Path 1abcdefgh
10from typing import ( 1abcdefgh
11 IO,
12 TYPE_CHECKING,
13 Any,
14 AnyStr,
15 Generic,
16 Literal,
17 TypedDict,
18 TypeVar,
19)
21from anyio import Path as AsyncPath 1abcdefgh
22from runtime_generics import runtime_generic, type_check 1abcdefgh
24from configzen.data import DataFormat 1abcdefgh
26if TYPE_CHECKING: 1abcdefgh
27 from collections.abc import Callable
28 from typing import ClassVar, overload
30 from typing_extensions import Never, Unpack
32 from configzen.data import Data
34 # We actually depend on `option_name` of every data format at runtime,
35 # but this time we pretend it doesn't exist.
36 from configzen.formats.std_json import JSONOptions
37 from configzen.formats.std_plist import PlistOptions
38 from configzen.formats.toml import TOMLOptions
39 from configzen.formats.yaml import YAMLOptions
41 class FormatOptions(TypedDict, total=False):
42 json: JSONOptions
43 plist: PlistOptions
44 toml: TOMLOptions
45 yaml: YAMLOptions
48__all__ = ( 1abcdefgh
49 "ConfigSource",
50 "FileConfigSource",
51 "StreamConfigSource",
52 "get_config_source",
53 "get_stream_config_source",
54 "get_file_config_source",
55)
57SourceType = TypeVar("SourceType") 1abcdefgh
60@runtime_generic 1abcdefgh
61class ConfigSource(Generic[SourceType, AnyStr], metaclass=ABCMeta): 1abcdefgh
62 """
63 Core interface for loading and saving configuration data.
65 If you need to implement your own configuration source class,
66 implement a subclass of this class and pass in to the `.config_load()` method
67 of your configuration or its model_config.
68 """
70 # Set up temporary stream factories
71 _binary_stream_factory: ClassVar[Callable[..., IO[bytes]]] = BytesIO 1abcdefgh
72 _string_stream_factory: ClassVar[Callable[..., IO[str]]] = StringIO 1abcdefgh
73 _data_format: DataFormat[Any, AnyStr] 1abcdefgh
74 source: SourceType 1abcdefgh
75 options: FormatOptions 1abcdefgh
77 def __init__( 1abcdefg
78 self,
79 source: SourceType,
80 data_format: str | DataFormat[Any, AnyStr] | None = None,
81 **options: Unpack[FormatOptions],
82 ) -> None:
83 self._temp_stream_factory: Callable[..., IO[AnyStr]] = (
84 self._binary_stream_factory
85 if self.is_binary()
86 else self._string_stream_factory
87 )
88 self.source = source
89 self.options = options
90 self.data_format = data_format # type: ignore[assignment]
92 @property 1abcdefgh
93 def data_format(self) -> DataFormat[Any, AnyStr]: 1abcdefgh
94 """The current data format for a configuration source."""
95 return self._data_format
97 @data_format.setter 1abcdefgh
98 def data_format(self, data_format: str | DataFormat[Any, AnyStr] | None) -> None: 1abcdefgh
99 if data_format is None:
100 data_format = self._guess_data_format()
101 else:
102 data_format = self._make_data_format(data_format)
103 data_format.validate_source(self)
104 self._data_format = data_format
106 def _guess_data_format(self) -> DataFormat[Any, AnyStr]: 1abcdefgh
107 msg = "Cannot guess the data format of the configuration source"
108 raise NotImplementedError(msg)
110 def _make_data_format( 1abcdefgh
111 self,
112 data_format: str | DataFormat[Any, AnyStr],
113 ) -> DataFormat[Any, AnyStr]:
114 if isinstance(data_format, str):
115 return DataFormat.for_extension(
116 data_format,
117 self.options.get(data_format), # type: ignore[arg-type]
118 )
119 data_format.configure(**self.options) # type: ignore[misc]
120 return data_format
122 if TYPE_CHECKING: 1abcdefgh
123 # python/mypy#9937
124 @overload
125 def is_binary(self: ConfigSource[SourceType, str]) -> Literal[False]: ...
127 @overload
128 def is_binary(
129 self: ConfigSource[SourceType, bytes],
130 ) -> Literal[True]: ...
132 def is_binary(self: ConfigSource[SourceType, AnyStr]) -> bool: 1abcdefgh
133 """Determine whether the configuration source is binary."""
134 return not type_check(self, ConfigSource[Any, str])
136 @abstractmethod 1abcdefgh
137 def load(self) -> Data: 1abcdefgh
138 """
139 Load the configuration source.
141 Return its contents as a dictionary.
142 """
143 raise NotImplementedError
145 @abstractmethod 1abcdefgh
146 async def load_async(self) -> Data: 1abcdefgh
147 """
148 Load the configuration source asynchronously.
150 Return its contents as a dictionary.
151 """
152 raise NotImplementedError
154 @abstractmethod 1abcdefgh
155 def dump(self, data: Data) -> None: 1abcdefgh
156 """Dump the configuration source."""
157 raise NotImplementedError
159 @abstractmethod 1abcdefgh
160 async def dump_async(self, data: Data) -> int: 1abcdefgh
161 """Dump the configuration source asynchronously."""
162 raise NotImplementedError
165@singledispatch 1abcdefgh
166def get_config_source( 1abcdefg
167 source: object,
168 _data_format: DataFormat[Any, AnyStr] | None = None,
169) -> ConfigSource[Any, Any]:
170 """Get a dedicated interface for a configuration source."""
171 type_name = type(source).__name__
172 msg = (
173 f"There is no class operating on {type_name!r} configuration "
174 f"sources. Implement it by creating a subclass of ConfigSource."
175 )
176 raise NotImplementedError(msg)
179def _make_path( 1abcdefgh
180 source: str | bytes | PathLike[str] | PathLike[bytes],
181) -> Path:
182 if isinstance(source, PathLike):
183 source = source.__fspath__()
184 if isinstance(source, bytes):
185 source = source.decode()
186 return Path(source)
189@runtime_generic 1abcdefgh
190class StreamConfigSource( 1abcdefgh
191 Generic[AnyStr],
192 ConfigSource[IO[Any], Any],
193):
194 """
195 A configuration source that is a stream.
197 Parameters
198 ----------
199 source
200 The stream to the configuration source.
202 """
204 def __init__( 1abcdefgh
205 self,
206 source: IO[AnyStr],
207 data_format: str | DataFormat[Any, AnyStr],
208 **options: Unpack[FormatOptions],
209 ) -> None:
210 super().__init__(source, data_format=data_format, **options)
212 def load(self) -> Data: 1abcdefgh
213 """
214 Load the configuration source.
216 Return its contents as a dictionary.
217 """
218 return self.data_format.load(self.source)
220 def load_async(self) -> Never: 1abcdefgh
221 """Unsupported."""
222 msg = "async streams are not supported for `StreamConfigSource`"
223 raise NotImplementedError(msg)
225 def dump(self, data: Data) -> None: 1abcdefgh
226 """Dump the configuration source."""
227 self.data_format.dump(data, self.source)
229 def dump_async(self, _data: Data) -> Never: 1abcdefgh
230 """Unsupported."""
231 msg = "async streams are not supported for `StreamConfigSource`"
232 raise NotImplementedError(msg)
235@get_config_source.register(BytesIO) 1abcdefgh
236@get_config_source.register(StringIO) 1abcdefgh
237def get_stream_config_source( 1abcdefgh
238 source: IO[bytes] | IO[str],
239 data_format: DataFormat[Any, Any],
240) -> StreamConfigSource[str] | StreamConfigSource[bytes]:
241 """Get a dedicated interface for a configuration source stream."""
242 return StreamConfigSource(source, data_format=data_format)
245@runtime_generic 1abcdefgh
246class FileConfigSource( 1abcdefgh
247 Generic[AnyStr],
248 ConfigSource[Path, AnyStr],
249):
250 """
251 A configuration source that is a file.
253 Parameters
254 ----------
255 source
256 The path to the configuration source file.
258 """
260 def __init__( 1abcdefg
261 self,
262 source: str | bytes | PathLike[str] | PathLike[bytes],
263 data_format: str | DataFormat[Any, Any] | None = None,
264 *,
265 use_processing_trace: bool = True,
266 **options: Unpack[FormatOptions],
267 ) -> None:
268 super().__init__(_make_path(source), data_format=data_format, **options)
269 self._use_processing_trace = use_processing_trace
271 @property 1abcdefgh
272 def paths(self) -> list[Path]: 1abcdefgh
273 """List possible path variants basing on the processing context trace."""
274 from configzen.config import processing
276 if (
277 not self.source.is_absolute()
278 and self._use_processing_trace
279 and (processing_context := processing.get())
280 ):
281 return [
282 _make_path(source).parent / self.source
283 for config_source in processing_context.trace
284 if isinstance(source := config_source.source, (str, bytes, PathLike))
285 ]
286 return [self.source] # in current working dir
288 def _guess_data_format(self) -> DataFormat[Any, AnyStr]: 1abcdefgh
289 suffix = self.source.suffix
290 if suffix:
291 extension = suffix.replace(".", "", 1)
292 from configzen.data import DataFormat
294 data_format_class = DataFormat.extension_registry.get(extension)
295 if data_format_class is not None:
296 return data_format_class(
297 self.options.get(data_format_class.option_name) or {},
298 )
299 msg = (
300 f"Cannot guess the data format of the configuration source "
301 f"with extension {suffix!r}"
302 )
303 raise NotImplementedError(msg)
305 def _after_load(self) -> None: 1abcdefgh
306 from configzen.config import processing
308 processing_context = processing.get()
309 if processing_context:
310 processing_context.trace.append(self)
312 def load(self) -> Data: 1abcdefgh
313 """
314 Load the configuration source file.
316 Return its contents as a dictionary.
317 """
318 data = self.data_format.load(self._temp_stream_factory(self.read()))
319 self._after_load()
320 return data
322 async def load_async(self) -> Data: 1abcdefgh
323 """
324 Load the configuration source file asynchronously.
326 Return its contents as a dictionary.
327 """
328 data = self.data_format.load(self._temp_stream_factory(await self.read_async()))
329 self._after_load()
330 return data
332 def dump(self, data: Data) -> None: 1abcdefgh
333 """
334 Dump the configuration data to the source file.
336 Parameters
337 ----------
338 data
339 The data to dump to the configuration source.
341 """
342 temp_stream = self._temp_stream_factory()
343 self.data_format.dump(data, temp_stream)
344 temp_stream.seek(0)
345 self.write(temp_stream.read())
347 async def dump_async(self, data: Data) -> int: 1abcdefgh
348 """
349 Load the configuration source file asynchronously.
351 Return its contents as a dictionary.
353 Parameters
354 ----------
355 data
356 The data to dump to the configuration source.
358 """
359 temp_stream = self._temp_stream_factory()
360 self.data_format.dump(data, temp_stream)
361 temp_stream.seek(0)
362 return await self.write_async(temp_stream.read())
364 def read(self) -> AnyStr: 1abcdefgh
365 """Read the configuration source and return its contents."""
366 errors = []
367 reader = Path.read_bytes if self.is_binary() else Path.read_text
368 for path in self.paths:
369 try:
370 return reader(path)
371 except FileNotFoundError as e: # noqa: PERF203
372 errors.append(e)
373 continue
374 raise FileNotFoundError(errors)
376 async def read_async(self) -> AnyStr: 1abcdefgh
377 """Read the configuration source file asynchronously and return its contents."""
378 errors = []
379 reader = AsyncPath.read_bytes if self.is_binary() else AsyncPath.read_text
380 for path in map(AsyncPath, self.paths):
381 try:
382 return await reader(path)
383 except FileNotFoundError as e: # noqa: PERF203
384 errors.append(e)
385 continue
386 raise FileNotFoundError(errors)
388 def write(self, content: AnyStr) -> int: 1abcdefgh
389 """
390 Write the configuration source file and return the number of bytes written.
392 Parameters
393 ----------
394 content
395 The content to write to the configuration source.
397 """
398 if self.is_binary():
399 return self.source.write_bytes(content)
400 return self.source.write_text(content)
402 async def write_async(self, content: AnyStr) -> int: 1abcdefgh
403 """
404 Write the configuration source file asynchronously.
406 Return the number of bytes written.
408 Parameters
409 ----------
410 content
411 The content to write to the configuration source.
413 """
414 if self.is_binary():
415 return await AsyncPath(self.source).write_bytes(content)
416 return await AsyncPath(self.source).write_text(content)
419@get_config_source.register(str) 1abcdefgh
420@get_config_source.register(bytes) 1abcdefgh
421@get_config_source.register(PathLike) 1abcdefgh
422def get_file_config_source( 1abcdefg
423 source: str | bytes | PathLike[str] | PathLike[bytes],
424 data_format: DataFormat[Any, AnyStr] | None = None,
425) -> FileConfigSource[str] | FileConfigSource[bytes]:
426 """Get a dedicated interface for a configuration source file."""
427 return FileConfigSource(source, data_format=data_format)