Skip to content

climate_ref.results.artifacts #

Path resolution for execution output artifacts.

ArtifactsReader is reached via Reader.artifacts. It resolves the primitive fragments an execution already stores (output_fragment, Execution.path, ExecutionOutput.filename) into filesystem Paths under a results root directory.

This module is deliberately narrow: no DB, no Config, and no coupling to the executions DTOs. It depends on a single results root path -- not the whole Config -- passed straight in; a value object would only earn its keep once a second root (e.g. an archive root) is needed. Resolution is purely lexical over the results root; opening/streaming the resolved paths stays with the consumer.

ArtifactsReader #

Resolves execution output fragments into filesystem paths.

Constructed from a single results root directory. Every method is containment-guarded: a resolved path that would escape the results root raises ValueError rather than returning a path outside it.

Source code in packages/climate-ref/src/climate_ref/results/artifacts.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class ArtifactsReader:
    """
    Resolves execution output fragments into filesystem paths.

    Constructed from a single ``results`` root directory.
    Every method is containment-guarded: a resolved path that would escape the results root
    raises ``ValueError`` rather than returning a path outside it.
    """

    def __init__(self, results: Path) -> None:
        self._results = results

    def _within(self, base: str, *parts: str) -> Path:
        """
        Join ``parts`` onto ``base`` and guard against escaping ``base``.

        Lexically normalises the joined path (no filesystem access, no symlink resolution) and
        checks containment with ``os.path.commonpath`` -- a real containment primitive, not a
        string-prefix check (which would wrongly accept a sibling directory like ``results2``).
        """
        root = os.path.normpath(base)
        candidate = os.path.normpath(os.path.join(root, *parts))
        if os.path.commonpath([root, candidate]) != root:
            raise ValueError(f"Path {candidate!r} escapes {root!r}")
        return Path(candidate)

    def _within_results(self, *parts: str) -> Path:
        """Join ``parts`` onto the results root and guard against escaping it."""
        return self._within(str(self._results), *parts)

    def output_directory(self, output_fragment: str) -> Path:
        """Resolve an execution's output directory from its ``output_fragment``."""
        return self._within_results(output_fragment)

    def log_file(self, output_fragment: str) -> Path:
        """Resolve the execution log file within an execution's output directory."""
        output_dir = self.output_directory(output_fragment)
        return self._within(str(output_dir), EXECUTION_LOG_FILENAME)

    def bundle(self, output_fragment: str, bundle_path: str | None) -> Path | None:
        """
        Resolve an execution's output bundle (``Execution.path``), if one is set.

        Returns ``None`` when ``bundle_path`` is ``None`` (no bundle recorded).
        """
        if bundle_path is None:
            return None
        output_dir = self.output_directory(output_fragment)
        return self._within(str(output_dir), bundle_path)

    def output_file(self, output_fragment: str, filename: str) -> Path:
        """Resolve a registered ``ExecutionOutput.filename`` within an execution's output directory."""
        output_dir = self.output_directory(output_fragment)
        return self._within(str(output_dir), filename)

bundle(output_fragment, bundle_path) #

Resolve an execution's output bundle (Execution.path), if one is set.

Returns None when bundle_path is None (no bundle recorded).

Source code in packages/climate-ref/src/climate_ref/results/artifacts.py 
63
64
65
66
67
68
69
70
71
72
def bundle(self, output_fragment: str, bundle_path: str | None) -> Path | None:
    """
    Resolve an execution's output bundle (``Execution.path``), if one is set.

    Returns ``None`` when ``bundle_path`` is ``None`` (no bundle recorded).
    """
    if bundle_path is None:
        return None
    output_dir = self.output_directory(output_fragment)
    return self._within(str(output_dir), bundle_path)

log_file(output_fragment) #

Resolve the execution log file within an execution's output directory.

Source code in packages/climate-ref/src/climate_ref/results/artifacts.py 
58
59
60
61
def log_file(self, output_fragment: str) -> Path:
    """Resolve the execution log file within an execution's output directory."""
    output_dir = self.output_directory(output_fragment)
    return self._within(str(output_dir), EXECUTION_LOG_FILENAME)

output_directory(output_fragment) #

Resolve an execution's output directory from its output_fragment.

Source code in packages/climate-ref/src/climate_ref/results/artifacts.py 
54
55
56
def output_directory(self, output_fragment: str) -> Path:
    """Resolve an execution's output directory from its ``output_fragment``."""
    return self._within_results(output_fragment)

output_file(output_fragment, filename) #

Resolve a registered ExecutionOutput.filename within an execution's output directory.

Source code in packages/climate-ref/src/climate_ref/results/artifacts.py 
74
75
76
77
def output_file(self, output_fragment: str, filename: str) -> Path:
    """Resolve a registered ``ExecutionOutput.filename`` within an execution's output directory."""
    output_dir = self.output_directory(output_fragment)
    return self._within(str(output_dir), filename)