Add loaded sources in ConfigSource.load()/ConfigSource.load_async()

configzen/config.py

@@ -62,7 +62,7 @@ class ProcessingContext(NamedTuple):
  model_class: type[BaseConfig]
  processor: ConfigProcessor
  # We keep it mainly to make path resolution smarter.
- trace: list[ConfigSource]
+ trace: list[ConfigSource[Any, Any]]
 
 
 def _locate(
@@ -326,10 +326,10 @@ def config_load(
  # than the configuration data, by using `processor.get_processed_data()`.
  processor = make_processor(config_source.load())
 
+ processing_context = ProcessingContext(cls, processor, trace=[config_source])
+ processing.set(processing_context)
  # ruff: noqa: FBT003
  try:
- processing.set(ProcessingContext(cls, processor, trace=[config_source]))
-
  # Processing will execute any commands that are present
  # in the configuration data and return the final configuration
  # data that we will use to construct an instance of the configuration model.
@@ -384,9 +384,9 @@ async def config_load_async(
  make_processor = cls._validate_processor_factory(processor_factory)
  processor = make_processor(await config_source.load_async())
 
+ processing_context = ProcessingContext(cls, processor, trace=[config_source])
+ processing.set(processing_context)
  try:
- processing.set(ProcessingContext(cls, processor, trace=[config_source]))
-
  self = cls(**await run_sync(processor.get_processed_data))
  finally:
  processing.set(None)

configzen/sources.py

@@ -302,21 +302,32 @@ def _guess_data_format(self) -> DataFormat[Any, AnyStr]:
  )
  raise NotImplementedError(msg)
 
+ def _after_load(self) -> None:
+ from configzen.config import processing
+
+ processing_context = processing.get()
+ if processing_context:
+ processing_context.trace.append(self)
+
  def load(self) -> Data:
  """
  Load the configuration source file.
 
  Return its contents as a dictionary.
  """
- return self.data_format.load(self._temp_stream_factory(self.read()))
+ data = self.data_format.load(self._temp_stream_factory(self.read()))
+ self._after_load()
+ return data
 
  async def load_async(self) -> Data:
  """
  Load the configuration source file asynchronously.
 
  Return its contents as a dictionary.
  """
- return self.data_format.load(self._temp_stream_factory(await self.read_async()))
+ data = self.data_format.load(self._temp_stream_factory(await self.read_async()))
+ self._after_load()
+ return data
 
  def dump(self, data: Data) -> None:
  """