Coverage for /builds/svesch/cikit/src/cikit/artifacts.py: 92%
120 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-09 14:17 +0000
1# SPDX-License-Identifier: GPL-3.0-or-later
2"""Artifact metadata and local registry support.
4This module provides the artifact model used by cikit together with a small
5local registry for persisting artifact metadata per logical run. Runtime
6artifact paths are absolute, while persisted paths are stored relative to a
7configurable artifact root for portability across workspaces.
9Author:
10 Sven Schlender
11Date:
12 2026-09-26
13"""
15import hashlib
16import logging
17from dataclasses import dataclass
18from enum import StrEnum
19from pathlib import Path
21from cikit.errors import CiToolsError
22from cikit.json_utils import load_json_file, save_json_file
24LOGGER = logging.getLogger(__name__)
25_ARTIFACT_FILE_SUFFIX = ".artifacts.json"
26_HASH_CHUNK_SIZE = 1024 * 1024
29class ArtifactType(StrEnum):
30 """Supported artifact types.
32 The enum values are persisted in JSON files and therefore must remain
33 stable across releases.
34 """
36 BADGE = "badge"
37 TIMELINE = "timeline"
38 XUNIT = "xunit"
39 LINT = "lint"
40 RST = "rst"
41 FOLDER = "folder"
42 METADATA = "meta"
43 EXTENSION = "extension"
44 LOG = "log"
45 BUILD = "build"
46 BUNDLE = "bundle"
49@dataclass(frozen=True)
50class Artifact:
51 """Describe a registered artifact.
53 Artifact paths are absolute so instances can be used directly for file
54 system access. Size and hash metadata are optional because directory
55 artifacts do not provide these values by default.
57 Args:
58 path: Absolute path identifying the artifact.
59 target_name: Final file or directory name of the artifact.
60 artifact_type: Type describing the artifact purpose.
61 run: Logical run associated with the artifact.
62 size: Optional artifact size in bytes.
63 hash: Optional artifact hash.
64 """
66 path: Path
67 target_name: str
68 artifact_type: ArtifactType
69 run: str
70 size: int = None
71 hash: str = None
73 def __post_init__(self):
74 """Validate and normalize the artifact path.
76 Raises:
77 CiToolsError: If the artifact path is not absolute.
78 """
79 if not self.path.is_absolute():
80 raise CiToolsError(f"Artifact path '{self.path}' must be absolute.")
82 object.__setattr__(self, "path", self.path.resolve())
85class ArtifactRegistry:
86 """Manage locally persisted artifact metadata.
88 Artifact metadata is stored in one JSON file per logical run below the
89 configured cache directory. Artifact paths are persisted relative to the
90 configured artifact root and restored as absolute paths when loaded.
92 The registry does not interpret the semantic meaning of the artifact root
93 and has no dependency on CI environment concepts.
94 """
96 def __init__(self, cache_dir: Path, artifact_root: Path):
97 """Initialize the artifact registry.
99 Args:
100 cache_dir: Directory used to store artifact metadata.
101 artifact_root: Base directory used to persist portable artifact paths.
102 """
103 self.cache_dir = cache_dir
104 self.artifact_root = artifact_root.resolve()
106 def add(self, artifact: Artifact) -> Artifact:
107 """Add or update an artifact entry.
109 Existing entries with the same persisted path in the same logical run
110 are overwritten. Size and hash metadata are stored together with the
111 basic artifact information.
113 Args:
114 artifact: Artifact metadata to store.
116 Returns:
117 Stored artifact.
119 Raises:
120 CiToolsError: If the artifact path lies outside ``artifact_root`` or
121 metadata cannot be read or written.
122 """
123 try:
124 relative_path = artifact.path.relative_to(self.artifact_root)
125 except ValueError as exp:
126 raise CiToolsError(f"Artifact path '{artifact.path}' is outside artifact root "
127 f"'{self.artifact_root}'.") from exp
129 filepath = _get_run_artifacts_filename(self.cache_dir, artifact.run)
130 artifacts = load_json_file(filepath, description="artifact data")
132 if not isinstance(artifacts, dict): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 artifacts = {}
135 artifacts[relative_path.as_posix()] = {
136 "target_name": artifact.target_name,
137 "type": artifact.artifact_type.value,
138 "size": artifact.size,
139 "hash": artifact.hash,
140 }
142 save_json_file(filepath, artifacts, description="artifact data")
143 return artifact
145 def _create_artifact(self, path: str, meta: dict, run_name: str,
146 artifact_type: ArtifactType) -> Artifact:
147 """Create an artifact from persisted registry metadata.
149 Args:
150 path: Persisted artifact path.
151 meta: Persisted artifact metadata.
152 run_name: Logical run associated with the artifact.
153 artifact_type: Resolved artifact type.
155 Returns:
156 Restored artifact instance.
158 Raises:
159 CiToolsError: If required artifact metadata is missing.
160 """
161 try:
162 target_name = meta["target_name"]
163 size = meta["size"]
164 artifact_hash = meta["hash"]
165 except KeyError as exp:
166 raise CiToolsError(f"Artifact entry '{path}' is incomplete.") from exp
168 artifact_path = self._resolve_artifact_path(Path(path))
170 return Artifact(artifact_path, target_name, artifact_type, run_name, size, artifact_hash)
172 def get(self, run: str = None, artifact_type: ArtifactType = None) -> list[Artifact]:
173 """Return locally stored artifacts.
175 Artifacts can optionally be filtered by logical run and artifact type.
176 If ``run`` is omitted, artifacts from all locally available runs are
177 returned.
179 Args:
180 run: Optional logical run name. If omitted, all runs are included.
181 artifact_type: Optional artifact type filter.
183 Returns:
184 Sorted list of matching artifacts with absolute paths.
186 Raises:
187 CiToolsError: If artifact metadata cannot be read or contains invalid
188 persisted data.
189 """
190 result = []
192 if run is not None:
193 artifact_files = [_get_run_artifacts_filename(self.cache_dir, run)]
194 else:
195 artifact_files = sorted(self.cache_dir.glob(f"*{_ARTIFACT_FILE_SUFFIX}"))
197 for filepath in artifact_files:
198 run_name = _get_run_name_from_path(filepath)
199 artifacts = load_json_file(filepath, description="artifact data")
201 if not isinstance(artifacts, dict): 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 continue
204 for path, meta in artifacts.items():
205 if not isinstance(meta, dict):
206 continue
208 type_value = meta.get("type", "")
210 try:
211 type_enum = ArtifactType(type_value)
212 except ValueError:
213 LOGGER.warning("Got unknown artifact type '%s'!", type_value)
214 continue
216 if artifact_type is not None and type_enum != artifact_type:
217 continue
219 result.append(self._create_artifact(path, meta, run_name, type_enum))
221 return sorted(result, key=lambda item: (item.run, item.path.as_posix()))
223 def _resolve_artifact_path(self, path: Path) -> Path:
224 """Resolve a persisted artifact path below the artifact root.
226 Args:
227 path: Persisted artifact path relative to ``artifact_root``.
229 Returns:
230 Absolute artifact path.
232 Raises:
233 CiToolsError: If the persisted path is absolute or escapes the
234 configured artifact root.
235 """
236 if path.is_absolute():
237 raise CiToolsError(f"Persisted artifact path '{path}' must be relative.")
239 artifact_path = (self.artifact_root / path).resolve()
241 try:
242 artifact_path.relative_to(self.artifact_root)
243 except ValueError as exp:
244 raise CiToolsError(f"Persisted artifact path '{path}' escapes artifact root "
245 f"'{self.artifact_root}'.") from exp
247 return artifact_path
250def calculate_artifact_hash(path: Path, artifact_type: ArtifactType) -> str:
251 """Calculate the SHA-256 hash of an artifact.
253 Folder artifacts do not provide a hash and therefore return ``None``.
254 File artifacts are processed incrementally to avoid loading large files
255 completely into memory.
257 Args:
258 path: Artifact path.
259 artifact_type: Type describing the artifact purpose.
261 Returns:
262 SHA-256 digest or ``None`` for folder artifacts.
264 Raises:
265 CiToolsError: If the file cannot be read.
266 """
267 if artifact_type == ArtifactType.FOLDER:
268 return None
270 digest = hashlib.sha256()
272 try:
273 with path.open("rb") as file:
274 for chunk in iter(lambda: file.read(_HASH_CHUNK_SIZE), b""):
275 digest.update(chunk)
276 except OSError as exp:
277 raise CiToolsError(f"Failed to calculate hash for artifact '{path}': {exp}") from exp
279 return digest.hexdigest()
282def get_artifact_size(path: Path, artifact_type: ArtifactType) -> int:
283 """Return the size of an artifact.
285 Folder artifacts do not provide a size and therefore return ``None``.
286 File artifacts return their size in bytes.
288 Args:
289 path: Artifact path.
290 artifact_type: Type describing the artifact purpose.
292 Returns:
293 File size in bytes or ``None`` for folder artifacts.
295 Raises:
296 CiToolsError: If the file metadata cannot be read.
297 """
298 if artifact_type == ArtifactType.FOLDER:
299 return None
301 try:
302 return path.stat().st_size
303 except OSError as exp:
304 raise CiToolsError(f"Failed to determine artifact size for '{path}': {exp}") from exp
307def _get_run_artifacts_filename(cache_dir: Path, run: str) -> Path:
308 """Return the artifact metadata file path for a run.
310 Args:
311 cache_dir: Base cache directory.
312 run: Logical run name.
314 Returns:
315 Full path to the run artifact metadata file.
316 """
317 return cache_dir / f"{run}{_ARTIFACT_FILE_SUFFIX}"
320def _get_run_name_from_path(filepath: Path) -> str:
321 """Extract the logical run name from an artifact metadata file.
323 Args:
324 filepath: Path to an artifact metadata file.
326 Returns:
327 Extracted logical run name.
329 Raises:
330 CiToolsError: If the file name does not contain a run name.
331 """
332 name = filepath.name
334 if name.endswith(_ARTIFACT_FILE_SUFFIX): 334 ↛ 337line 334 didn't jump to line 337 because the condition on line 334 was always true
335 run_name = name.removesuffix(_ARTIFACT_FILE_SUFFIX)
336 else:
337 run_name = filepath.stem
339 if not run_name: 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true
340 raise CiToolsError(f"Cannot extract run name from file '{filepath.name}'")
342 return run_name
345__all__ = ["Artifact", "ArtifactRegistry", "ArtifactType", "calculate_artifact_hash"]