Coverage for configzen/sources.py: 39%

1"""Sources and destinations that hold the configuration data.""" 

2 

3from __future__ import annotations 8 ctx1abcdefgh

4 

5from abc import ABCMeta, abstractmethod 8 ctx1abcdefgh

6from functools import singledispatch 8 ctx1abcdefgh

7from io import BytesIO, StringIO 8 ctx1abcdefgh

8from os import PathLike 8 ctx1abcdefgh

9from pathlib import Path 8 ctx1abcdefgh

10from typing import ( 8 ctx1abcdefgh

11 IO, 

12 TYPE_CHECKING, 

13 Any, 

14 AnyStr, 

15 Generic, 

16 Literal, 

17 TypedDict, 

18 TypeVar, 

19) 

20 

21from anyio import Path as AsyncPath 8 ctx1abcdefgh

22from runtime_generics import runtime_generic, type_check 8 ctx1abcdefgh

23 

24from configzen.data import DataFormat 8 ctx1abcdefgh

25 

26if TYPE_CHECKING: 8 ctx1abcdefgh

27 from collections.abc import Callable 

28 from typing import ClassVar, overload 

29 

30 from typing_extensions import Never, Unpack 

31 

32 from configzen.data import Data 

33 

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 

40 

41 class FormatOptions(TypedDict, total=False): 

42 json: JSONOptions 

43 plist: PlistOptions 

44 toml: TOMLOptions 

45 yaml: YAMLOptions 

46 

47 

48__all__ = ( 8 ctx1abcdefgh

49 "ConfigSource", 

50 "FileConfigSource", 

51 "StreamConfigSource", 

52 "get_config_source", 

53 "get_stream_config_source", 

54 "get_file_config_source", 

55) 

56 

57SourceType = TypeVar("SourceType") 8 ctx1abcdefgh

58 

59 

60@runtime_generic 8 ctx1abcdefgh

61class ConfigSource(Generic[SourceType, AnyStr], metaclass=ABCMeta): 8 ctx1abcdefgh

62 """ 

63 Core interface for loading and saving configuration data. 

64 

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 """ 

69 

70 # Set up temporary stream factories 

71 _binary_stream_factory: ClassVar[Callable[..., IO[bytes]]] = BytesIO 8 ctx1abcdefgh

72 _string_stream_factory: ClassVar[Callable[..., IO[str]]] = StringIO 8 ctx1abcdefgh

73 _data_format: DataFormat[Any, AnyStr] 8 ctx1abcdefgh

74 source: SourceType 8 ctx1abcdefgh

75 options: FormatOptions 8 ctx1abcdefgh

76 

77 def __init__( 7 ctx1abcdefg

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] 

91 

92 @property 8 ctx1abcdefgh

93 def data_format(self) -> DataFormat[Any, AnyStr]: 8 ctx1abcdefgh

94 """The current data format for a configuration source.""" 

95 return self._data_format 

96 

97 @data_format.setter 8 ctx1abcdefgh

98 def data_format(self, data_format: str | DataFormat[Any, AnyStr] | None) -> None: 8 ctx1abcdefgh

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 

105 

106 def _guess_data_format(self) -> DataFormat[Any, AnyStr]: 8 ctx1abcdefgh

107 msg = "Cannot guess the data format of the configuration source" 

108 raise NotImplementedError(msg) 

109 

110 def _make_data_format( 8 ctx1abcdefgh

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 

121 

122 if TYPE_CHECKING: 8 ctx1abcdefgh

123 # python/mypy#9937 

124 @overload 

125 def is_binary(self: ConfigSource[SourceType, str]) -> Literal[False]: ... 

126 

127 @overload 

128 def is_binary( 

129 self: ConfigSource[SourceType, bytes], 

130 ) -> Literal[True]: ... 

131 

132 def is_binary(self: ConfigSource[SourceType, AnyStr]) -> bool: 8 ctx1abcdefgh

133 """Determine whether the configuration source is binary.""" 

134 return not type_check(self, ConfigSource[Any, str]) 

135 

136 @abstractmethod 8 ctx1abcdefgh

137 def load(self) -> Data: 8 ctx1abcdefgh

138 """ 

139 Load the configuration source. 

140 

141 Return its contents as a dictionary. 

142 """ 

143 raise NotImplementedError 

144 

145 @abstractmethod 8 ctx1abcdefgh

146 async def load_async(self) -> Data: 8 ctx1abcdefgh

147 """ 

148 Load the configuration source asynchronously. 

149 

150 Return its contents as a dictionary. 

151 """ 

152 raise NotImplementedError 

153 

154 @abstractmethod 8 ctx1abcdefgh

155 def dump(self, data: Data) -> None: 8 ctx1abcdefgh

156 """Dump the configuration source.""" 

157 raise NotImplementedError 

158 

159 @abstractmethod 8 ctx1abcdefgh

160 async def dump_async(self, data: Data) -> int: 8 ctx1abcdefgh

161 """Dump the configuration source asynchronously.""" 

162 raise NotImplementedError 

163 

164 

165@singledispatch 8 ctx1abcdefgh

166def get_config_source( 7 ctx1abcdefg

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) 

177 

178 

179def _make_path( 8 ctx1abcdefgh

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) 

187 

188 

189@runtime_generic 8 ctx1abcdefgh

190class StreamConfigSource( 8 ctx1abcdefgh

191 Generic[AnyStr], 

192 ConfigSource[IO[Any], Any], 

193): 

194 """ 

195 A configuration source that is a stream. 

196 

197 Parameters 

198 ---------- 

199 source 

200 The stream to the configuration source. 

201 

202 """ 

203 

204 def __init__( 8 ctx1abcdefgh

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) 

211 

212 def load(self) -> Data: 8 ctx1abcdefgh

213 """ 

214 Load the configuration source. 

215 

216 Return its contents as a dictionary. 

217 """ 

218 return self.data_format.load(self.source) 

219 

220 def load_async(self) -> Never: 8 ctx1abcdefgh

221 """Unsupported.""" 

222 msg = "async streams are not supported for `StreamConfigSource`" 

223 raise NotImplementedError(msg) 

224 

225 def dump(self, data: Data) -> None: 8 ctx1abcdefgh

226 """Dump the configuration source.""" 

227 self.data_format.dump(data, self.source) 

228 

229 def dump_async(self, _data: Data) -> Never: 8 ctx1abcdefgh

230 """Unsupported.""" 

231 msg = "async streams are not supported for `StreamConfigSource`" 

232 raise NotImplementedError(msg) 

233 

234 

235@get_config_source.register(BytesIO) 8 ctx1abcdefgh

236@get_config_source.register(StringIO) 8 ctx1abcdefgh

237def get_stream_config_source( 8 ctx1abcdefgh

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) 

243 

244 

245@runtime_generic 8 ctx1abcdefgh

246class FileConfigSource( 8 ctx1abcdefgh

247 Generic[AnyStr], 

248 ConfigSource[Path, AnyStr], 

249): 

250 """ 

251 A configuration source that is a file. 

252 

253 Parameters 

254 ---------- 

255 source 

256 The path to the configuration source file. 

257 

258 """ 

259 

260 def __init__( 7 ctx1abcdefg

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 

270 

271 @property 8 ctx1abcdefgh

272 def paths(self) -> list[Path]: 8 ctx1abcdefgh

273 """List possible path variants basing on the processing context trace.""" 

274 from configzen.config import processing 

275 

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 

287 

288 def _guess_data_format(self) -> DataFormat[Any, AnyStr]: 8 ctx1abcdefgh

289 suffix = self.source.suffix 

290 if suffix: 

291 extension = suffix.replace(".", "", 1) 

292 from configzen.data import DataFormat 

293 

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) 

304 

305 def _after_load(self) -> None: 8 ctx1abcdefgh

306 from configzen.config import processing 

307 

308 processing_context = processing.get() 

309 if processing_context: 

310 processing_context.trace.append(self) 

311 

312 def load(self) -> Data: 8 ctx1abcdefgh

313 """ 

314 Load the configuration source file. 

315 

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 

321 

322 async def load_async(self) -> Data: 8 ctx1abcdefgh

323 """ 

324 Load the configuration source file asynchronously. 

325 

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 

331 

332 def dump(self, data: Data) -> None: 8 ctx1abcdefgh

333 """ 

334 Dump the configuration data to the source file. 

335 

336 Parameters 

337 ---------- 

338 data 

339 The data to dump to the configuration source. 

340 

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()) 

346 

347 async def dump_async(self, data: Data) -> int: 8 ctx1abcdefgh

348 """ 

349 Load the configuration source file asynchronously. 

350 

351 Return its contents as a dictionary. 

352 

353 Parameters 

354 ---------- 

355 data 

356 The data to dump to the configuration source. 

357 

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()) 

363 

364 def read(self) -> AnyStr: 8 ctx1abcdefgh

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) 

375 

376 async def read_async(self) -> AnyStr: 8 ctx1abcdefgh

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) 

387 

388 def write(self, content: AnyStr) -> int: 8 ctx1abcdefgh

389 """ 

390 Write the configuration source file and return the number of bytes written. 

391 

392 Parameters 

393 ---------- 

394 content 

395 The content to write to the configuration source. 

396 

397 """ 

398 if self.is_binary(): 

399 return self.source.write_bytes(content) 

400 return self.source.write_text(content) 

401 

402 async def write_async(self, content: AnyStr) -> int: 8 ctx1abcdefgh

403 """ 

404 Write the configuration source file asynchronously. 

405 

406 Return the number of bytes written. 

407 

408 Parameters 

409 ---------- 

410 content 

411 The content to write to the configuration source. 

412 

413 """ 

414 if self.is_binary(): 

415 return await AsyncPath(self.source).write_bytes(content) 

416 return await AsyncPath(self.source).write_text(content) 

417 

418 

419@get_config_source.register(str) 8 ctx1abcdefgh

420@get_config_source.register(bytes) 8 ctx1abcdefgh

421@get_config_source.register(PathLike) 8 ctx1abcdefgh

422def get_file_config_source( 7 ctx1abcdefg

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)