Coverage for /builds/svesch/cikit/src/cikit/env/api.py: 95%
230 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"""Public API for env.
4This module exposes CiEnvironment, which resolves a set of EnvVar definitions
5against the current OS environment. It also provides small helper functions
6useful for consumers, such as preparing directories, producing shell exports,
7and printing resolved environment values.
9Author:
10 Sven Schlender
11Date:
12 2026-02-18
13"""
15import logging
16import os
17import shlex
18import shutil
19from pathlib import Path
20from string import Template, hexdigits
21from typing import Any, Iterable
23from cikit.data import RefKey
24from cikit.artifacts import (Artifact, ArtifactRegistry, ArtifactType, get_artifact_size,
25 calculate_artifact_hash)
26from cikit.errors import CiToolsError, ResolvePathError
27from cikit.factory import build_run_context_from_env
28from cikit.command import CommandArgv
30from .provider import BaseProvider, detect_provider
33LOGGER = logging.getLogger()
34MASKED = "***"
35CLEANUP_DIRS = ("${PUBLIC_DIR}",)
36CREATE_DIRS = ("${PUBLIC_DIR}", "${CIKIT_DATA_DIR}",)
39_ARTIFACT_TARGET_DIRS = {
40 ArtifactType.BADGE: Path("badges"),
41 ArtifactType.RST: Path("rst"),
42 ArtifactType.TIMELINE: Path("artifacts"),
43 ArtifactType.XUNIT: Path("artifacts"),
44 ArtifactType.LINT: Path("artifacts"),
45 ArtifactType.FOLDER: Path("artifacts"),
46 ArtifactType.METADATA: Path("artifacts"),
47 ArtifactType.EXTENSION: Path("artifacts"),
48 ArtifactType.LOG: Path("artifacts"),
49 ArtifactType.BUILD: Path("artifacts"),
50 ArtifactType.BUNDLE: Path("artifacts"),
51}
54def _artifact_target_dir(public_dir: Path, artifact_type: ArtifactType) -> Path:
55 """Return the public target directory for an artifact type.
57 Artifact types are mapped to fixed subdirectories below ``PUBLIC_DIR``.
58 This keeps the persistent artifact layout independent from producer-specific
59 output directories such as temporary badge or report locations.
61 Args:
62 public_dir: Base directory for persistent public artifacts.
63 artifact_type: Artifact type used to select the target subdirectory.
65 Returns:
66 Target directory below ``public_dir``.
68 Raises:
69 CiToolsError: If no target directory is configured for the artifact type.
70 """
71 try:
72 relative_dir = _ARTIFACT_TARGET_DIRS[artifact_type]
73 except KeyError as exp:
74 raise CiToolsError("No public target directory configured for artifact type "
75 f"'{artifact_type}'.") from exp
77 return public_dir / relative_dir
80def write_file(output_file: Path, data: str):
81 """Write text data to a file using UTF-8 encoding.
83 The function writes the provided string to the given file path.
84 Any existing file will be overwritten. Errors during writing
85 are wrapped into a CiToolsError for consistent error handling.
87 Args:
88 output_file: Target file path to write to.
89 data: Text content to write into the file.
91 Raises:
92 CiToolsError: If writing the file fails due to an OS error.
93 """
94 try:
95 output_file.write_text(data, encoding="utf-8")
96 except OSError as exp:
97 raise CiToolsError(f"Error while writing to '{output_file}': {exp}") from exp
99 LOGGER.debug("Written %s", output_file)
102class CiEnvironment:
103 """Resolved CI environment with provider metadata access.
105 This class resolves a set of EnvVar definitions in declaration order and
106 stores the resulting values as a stable environment snapshot. Variables
107 may reference previously resolved variables via ``${VAR}`` templates.
109 In addition to resolved environment values, the environment exposes
110 provider-specific metadata such as pipeline identifiers, branch names
111 and commit information through dedicated properties.
113 The class also provides common helpers for path resolution, command
114 argument expansion and environment export.
115 """
117 def __init__(self, variables: list, *, provider: BaseProvider = None, cwd: str | Path = None):
118 """Resolve environment variables in definition order.
120 Variables are resolved sequentially in the order they appear in
121 ``variables``. Each variable may reference previously resolved
122 variables via ${VAR} template syntax.
124 Args:
125 variables: Ordered list of ``EnvVar`` definitions to resolve.
126 provider: Optional CI provider instance. If ``None``, the provider
127 is automatically detected.
128 cwd: Optional default working directory for command execution.
129 If omitted, ``PROJECT_ROOT`` is used.
131 Raises:
132 CiToolsError: If duplicate variables are configured or required
133 values cannot be resolved.
134 """
135 self.provider = provider if provider is not None else detect_provider()
137 vars_by_key = {}
138 for var in variables:
139 if var.key in vars_by_key:
140 raise CiToolsError(f"Duplicate EnvVar definition: {var.key}")
141 vars_by_key[var.key] = var
143 self.variables = vars_by_key
144 self.values = {}
146 os_env = dict(os.environ)
147 for var in vars_by_key.values():
148 self.values[var.key] = var.resolve(os_env, self.values)
150 self.cwd = Path(cwd or self["PROJECT_ROOT"]).resolve()
152 self._warn_relative_paths()
154 def _warn_relative_paths(self):
155 """Log a single warning for all resolved relative Path values.
157 After all environment variables have been resolved, this helper scans
158 the final value map for Path objects that are still relative. Relative
159 paths may lead to side effects that depend on the current working
160 directory.
161 """
162 relative = []
164 for key, value in self.values.items():
165 if isinstance(value, Path) and not value.is_absolute(): 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 relative.append((key, value))
168 if not relative: 168 ↛ 171line 168 didn't jump to line 171 because the condition on line 168 was always true
169 return
171 lines = [
172 "Resolved environment contains relative paths.",
173 "CI scripts are most reliable with absolute paths.",
174 "Relative path variables:",
175 ]
176 lines.extend(f" - {key}: {value}" for key, value in sorted(relative))
177 LOGGER.warning("\n".join(lines))
179 def _resolve_required(self, key: str) -> Any:
180 """Return the resolved value for a required environment variable.
182 This helper only accesses values that were resolved during
183 ``CiEnvironment`` initialization. It does not consult the provider
184 again and therefore treats the environment as a closed value set.
186 Args:
187 key: Environment variable name.
189 Returns:
190 Resolved value for ``key``.
192 Raises:
193 CiToolsError: If ``key`` is not present in the resolved environment.
194 """
195 try:
196 return self.values[key]
197 except KeyError as exp:
198 raise CiToolsError(f"Missing required environment variable: {key}") from exp
200 def __getitem__(self, key: str) -> Any:
201 """Return the resolved environment value for ``key``.
203 This accessor is intended for required variables and delegates to
204 ``_resolve_required()``. Missing keys are reported as ``CiToolsError``
205 instead of exposing a raw ``KeyError``.
207 Args:
208 key: Environment variable name.
210 Returns:
211 Resolved value associated with ``key``.
213 Raises:
214 CiToolsError: If ``key`` is not present in the resolved environment.
215 """
216 return self._resolve_required(key)
218 def get(self, key: str, default: Any = "") -> Any:
219 """Return the resolved environment value for ``key`` or default.
221 This method returns ``default`` when the requested key is missing.
222 It also returns ``default`` when the resolved value is ``None``.
224 Args:
225 key: Environment variable name.
226 default: Value returned if ``key`` is missing or resolved to ``None``.
228 Returns:
229 Resolved value for ``key`` or ``default``.
230 """
231 try:
232 value = self._resolve_required(key)
233 except CiToolsError:
234 return default
236 if value is None:
237 return default
239 return value
241 def as_dict(self, secrets: bool = False) -> dict[str, Any]:
242 """Return resolved values as a dictionary.
244 Args:
245 secrets: If False, secret values are replaced by ``MASKED``.
247 Returns:
248 Mapping of variable keys to resolved values.
249 """
250 return {var.key: (self.values[var.key] if secrets or not var.secret else MASKED)
251 for var in self.variables.values()}
253 def as_str_map(self, secrets: bool = True) -> dict[str, str]:
254 """Return resolved environment values as a string-only mapping.
256 This method converts all resolved values to strings and replaces
257 ``None`` with an empty string. It is intended for command execution
258 and template expansion.
260 Args:
261 secrets: If True, include secret values.
263 Returns:
264 Mapping of environment variable names to string values.
265 """
266 values = self.as_dict(secrets=secrets)
267 out = {}
269 for key, val in values.items():
270 out[key] = "" if val is None else str(val)
272 return out
274 def argv(self, cmd, subcommand: str = None, extra_args: list = None,
275 no_common: bool = False) -> CommandArgv:
276 """Build environment-aware command arguments.
278 The command argv is built via ``Command.argv()`` and resolved twice:
279 once with all environment values for process execution and once with
280 secret values masked for safe display.
282 If enabled on the Command, ``EXEC_TOOL`` is prefixed to both argument
283 representations.
285 Args:
286 cmd: Command definition used to build argv.
287 subcommand: Optional subcommand name.
288 extra_args: Optional additional command arguments.
289 no_common: If True, do not include command common arguments.
291 Returns:
292 Resolved command arguments for execution and safe display.
293 """
294 argv = cmd.argv(subcommand, extra_args, no_common)
296 exec_values = self.as_str_map(secrets=True)
297 masked_values = self.as_str_map(secrets=False)
299 exec_argv = [Template(str(arg)).safe_substitute(exec_values) for arg in argv]
300 masked_argv = [Template(str(arg)).safe_substitute(masked_values) for arg in argv]
302 if cmd.use_exec_tool:
303 prefix = self.get("EXEC_TOOL")
304 if prefix: 304 ↛ 309line 304 didn't jump to line 309 because the condition on line 304 was always true
305 prefix_argv = shlex.split(str(prefix))
306 exec_argv = prefix_argv + exec_argv
307 masked_argv = prefix_argv + masked_argv
309 return CommandArgv(exec_argv, masked_argv)
311 def resolve_path(self, path) -> Path:
312 """Resolve an environment-expanded path relative to PROJECT_ROOT.
314 Args:
315 path: Absolute path, relative path, or path template.
317 Returns:
318 Resolved absolute path.
320 Raises:
321 ResolvePathError: If a referenced environment variable is unknown.
322 """
323 try:
324 resolved = Template(str(path)).substitute(self.as_str_map(secrets=True))
325 except KeyError as exp:
326 raise ResolvePathError(f"Unresolved path template variable: {exp}") from exp
328 resolved_path = Path(resolved)
330 if resolved_path.is_absolute():
331 return resolved_path
333 return self["PROJECT_ROOT"] / resolved_path
335 @property
336 def pipeline_id(self) -> str:
337 """Return the provider-specific pipeline identifier.
339 Returns:
340 Provider-specific pipeline identifier.
341 """
342 return self.provider.pipeline_id
344 @property
345 def pipeline_alias(self) -> str:
346 """Return the provider-specific pipeline alias.
348 Returns:
349 Provider-specific human-readable pipeline identifier.
350 """
351 return self.provider.pipeline_alias
353 @property
354 def branch_name(self) -> str:
355 """Return the active branch name.
357 Returns:
358 Active branch name.
359 """
360 return self.provider.branch_name
362 @property
363 def commit_sha(self) -> str:
364 """Return the full commit SHA.
366 Returns:
367 Full commit SHA.
368 """
369 return self.provider.commit_sha
371 @property
372 def commit_short_sha(self) -> str:
373 """Return the short commit SHA.
375 Returns:
376 Short commit SHA.
377 """
378 return self.provider.commit_short_sha
380 @property
381 def commit_title(self) -> str:
382 """Return the commit title.
384 Returns:
385 Commit title.
386 """
387 return self.provider.commit_title
389 @property
390 def commit_message(self) -> str:
391 """Return the full commit message.
393 Returns:
394 Full commit message.
395 """
396 return self.provider.commit_message
398 @property
399 def commit_author(self) -> str:
400 """Return the commit author.
402 Returns:
403 Commit author.
404 """
405 return self.provider.commit_author
407 @property
408 def commit_timestamp(self) -> str:
409 """Return the commit timestamp.
411 Returns:
412 Commit timestamp.
413 """
414 return self.provider.commit_timestamp
417class CiRunEnvironment(CiEnvironment):
418 """CI environment with run data and artifact support.
420 This class extends ``CiEnvironment`` with a run-bound ``RunContext`` for
421 cikit data operations and a local artifact registry. It provides helpers
422 for updating run metrics, managing artifacts, and querying timelines while
423 reusing all generic environment functionality.
424 """
426 def __init__(self, variables: list, *, provider: BaseProvider = None, run_name: str,
427 cwd: str | Path = None):
428 """Initialize a CI environment with run data and artifact support.
430 This environment extends the base ``CiEnvironment`` by binding a
431 ``RunContext`` to a logical run and creating an ``ArtifactRegistry`` for
432 persisted artifact metadata. The run context provides access to cikit data
433 operations, while the artifact registry resolves persisted artifact paths
434 relative to ``PUBLIC_DIR``.
436 Args:
437 variables: List of environment variable definitions to resolve.
438 provider: Optional CI provider used for provider metadata.
439 run_name: Logical name of the run used to create and identify the
440 associated run context.
441 cwd: Optional default working directory for command execution.
442 If omitted, ``PROJECT_ROOT`` is used.
444 Raises:
445 CiToolsError: If the run context or artifact registry cannot be created from the
446 resolved environment.
447 """
448 super().__init__(variables, provider=provider, cwd=cwd)
450 self.run = build_run_context_from_env(self, run_name)
451 self._artifacts = ArtifactRegistry(self["CIKIT_DATA_DIR"], self["PUBLIC_DIR"])
453 @property
454 def name(self) -> str:
455 """Return the logical run name.
457 Returns:
458 Logical run name bound to this environment.
459 """
460 return self.run.run_name
462 def data_set(self, key: str, value):
463 """Set a single cikit data entry for the active run.
465 Stores a key-value pair for the active run. If the key already exists,
466 its value is overwritten and persisted through the configured cikit
467 data backend.
469 Args:
470 key: Name of the data field.
471 value: Value to store for the given key.
473 Returns:
474 Parsed backend response or merged local run data.
475 """
476 return self.run.set(key, value)
478 def data_update(self, data: dict):
479 """Merge cikit data values and persist them immediately.
481 The provided mapping is merged into the current run data state.
482 Existing keys are overwritten. The update is persisted through the
483 configured cikit data backend.
485 Args:
486 data: Mapping containing data fields to update.
488 Returns:
489 Parsed backend response or merged local run data.
490 """
491 return self.run.update(data)
493 @property
494 def data(self) -> dict:
495 """Return a copy of the current run data mapping.
497 The returned dictionary is a shallow copy of the locally cached run
498 data. Modifying the returned mapping does not affect the stored data
499 and does not trigger a backend update.
501 Returns:
502 Copy of the current run data mapping.
503 """
504 return self.run.get_data()
506 def copy_artifact(self, path: Path, artifact_type: ArtifactType,
507 target_name: str = None, artifact_hash: str = None) -> Artifact:
508 """Copy an artifact to its public target directory and register it.
510 Relative source paths are resolved below ``PROJECT_ROOT``. Absolute source
511 paths may point to any accessible location, including temporary directories.
512 Artifact size and hash metadata are determined after copying. A supplied
513 hash is preserved, otherwise a SHA-256 hash is calculated automatically.
515 Args:
516 path: Source artifact path, absolute or relative to ``PROJECT_ROOT``.
517 artifact_type: Type used to select the public target directory.
518 target_name: Optional target file or directory name. If omitted, the
519 source name is preserved.
520 artifact_hash: Optional precomputed SHA-256 hash.
522 Returns:
523 Registered artifact.
525 Raises:
526 CiToolsError: If the source does not exist, the supplied hash is not a valid
527 SHA-256 digest, or the artifact cannot be copied, inspected, hashed, or
528 registered.
529 """
530 project_root = self["PROJECT_ROOT"]
531 public_dir = self["PUBLIC_DIR"]
533 source = path if path.is_absolute() else project_root / path
534 source = source.resolve()
536 if not source.exists():
537 raise CiToolsError(f"Artifact path '{source}' does not exist.")
539 if artifact_hash is not None:
540 if len(artifact_hash) != 64 or any(char not in hexdigits for char in artifact_hash):
541 raise CiToolsError("Invalid artifact hash, expected a 64-character SHA-256 digest.")
543 artifact_hash = artifact_hash.lower()
545 target_dir = _artifact_target_dir(public_dir, artifact_type)
546 target_dir.mkdir(parents=True, exist_ok=True)
548 target = target_dir / (target_name or source.name)
550 try:
551 if source.is_dir():
552 shutil.copytree(source, target, dirs_exist_ok=True)
553 else:
554 target.parent.mkdir(parents=True, exist_ok=True)
555 shutil.copy2(source, target)
556 except OSError as exp:
557 raise CiToolsError(f"Failed to copy artifact '{source}' to '{target}': {exp}") from exp
559 size = get_artifact_size(target, artifact_type)
561 if artifact_hash is None:
562 artifact_hash = calculate_artifact_hash(target, artifact_type)
564 artifact = Artifact(target, target.name, artifact_type, self.name, size, artifact_hash)
565 self._artifacts.add(artifact)
567 LOGGER.info("Registered %s artifact '%s'", artifact_type.value, target.name)
569 return artifact
571 def get_artifacts(self, run: str = None, artifact_type: ArtifactType = None) -> list[Artifact]:
572 """Return registered artifacts.
574 Args:
575 run: Optional logical run name. If omitted, artifacts from all runs
576 are returned.
577 artifact_type: Optional artifact type filter.
579 Returns:
580 Sorted list of matching artifacts.
582 Raises:
583 CiToolsError: If artifact metadata cannot be read.
584 """
585 return self._artifacts.get(run, artifact_type)
587 def get_run_artifacts(self, artifact_type: ArtifactType = None) -> list[Artifact]:
588 """Return artifacts registered for the active run.
590 This is a convenience wrapper around ``get_artifacts()`` that automatically
591 uses the logical run bound to this ``CiRunEnvironment``. It is intended for
592 callers that operate on the current run and therefore should not need to
593 pass ``env.name`` explicitly.
595 Args:
596 artifact_type: Optional artifact type filter. If omitted, all artifact
597 types registered for the active run are returned.
599 Returns:
600 Sorted list of artifacts registered for the active run.
602 Raises:
603 CiToolsError: If artifact metadata cannot be read.
604 """
605 return self.get_artifacts(self.name, artifact_type)
607 def get_artifact(self, target_name: str, run: str = None,
608 artifact_type: ArtifactType = None) -> Artifact:
609 """Return one registered artifact.
611 Args:
612 target_name: Registered artifact file or directory name.
613 run: Optional logical run name. If omitted, all runs are searched.
614 artifact_type: Optional artifact type used to restrict the search.
616 Returns:
617 Matching artifact.
619 Raises:
620 CiToolsError: If no matching artifact is found, the result is ambiguous,
621 or artifact metadata cannot be read.
622 """
623 artifacts = self.get_artifacts(run, artifact_type)
624 matches = [artifact for artifact in artifacts if artifact.target_name == target_name]
626 if not matches:
627 raise CiToolsError(f"Artifact '{target_name}' not found.")
629 if len(matches) > 1:
630 raise CiToolsError(f"Artifact '{target_name}' is ambiguous.")
632 return matches[0]
634 def get_run_artifact(self, target_name: str, artifact_type: ArtifactType = None) -> Artifact:
635 """Return one artifact registered for the active run.
637 This is a convenience wrapper around ``get_artifact()`` that restricts the
638 lookup to the logical run bound to this ``CiRunEnvironment``. It avoids
639 repeating ``env.name`` at call sites while preserving the general
640 ``get_artifact()`` behavior where ``run=None`` searches across all runs.
642 Args:
643 target_name: Registered artifact file or directory name.
644 artifact_type: Optional artifact type used to restrict the lookup.
646 Returns:
647 Matching artifact registered for the active run.
649 Raises:
650 CiToolsError: If no matching artifact is found, the result is
651 ambiguous, or artifact metadata cannot be read.
652 """
653 return self.get_artifact(target_name, self.name, artifact_type)
655 def timeline(self, keys: list[str], ref_key: RefKey = RefKey.INDEX_DESC,
656 limit: int = None) -> dict:
657 """Return timeline data for multiple keys of the active run.
659 This convenience wrapper queries the bound RunContext for the given
660 cikit data keys of the current logical run. The returned structure can
661 be used directly for chart generation.
663 Args:
664 keys: List of cikit data keys to query.
665 ref_key: Reference key used for the X-axis.
666 limit: Optional maximum number of timeline entries returned by the API.
668 Returns:
669 Parsed timeline response containing ``refs`` and ``series``.
671 Raises:
672 CiRecoverableError: If no remote cikit data access is configured.
673 CiToolsError: If the timeline query fails.
674 """
675 return self.run.timeline(keys, ref_key, limit)
678def prepare_dirs(env: CiEnvironment):
679 """Create required output directories and clean selected directories.
681 Args:
682 env: Resolved CI environment.
683 """
684 for directory in CLEANUP_DIRS:
685 try:
686 path = env.resolve_path(directory)
687 except ResolvePathError as exp:
688 LOGGER.warning("Skipping cleanup directory: %s", exp)
689 continue
691 if path.exists():
692 LOGGER.info("Removing dir '%s'", path)
693 shutil.rmtree(path)
695 for directory in CREATE_DIRS:
696 try:
697 path = env.resolve_path(directory)
698 except ResolvePathError as exp:
699 LOGGER.warning("Skipping create directory: %s", exp)
700 continue
702 path.mkdir(parents=True, exist_ok=True)
705def shell_export_lines(env: CiEnvironment, secrets: bool = False) -> Iterable[str]:
706 """Yield shell export lines for resolved environment variables.
708 Args:
709 env: Resolved CI environment.
710 secrets: If True, include secret values.
712 Yields:
713 Export lines suitable for shell evaluation.
714 """
715 for var in env.variables.values():
716 if var.secret and not secrets:
717 continue
719 val = env.values.get(var.key)
720 if val is None: 720 ↛ 721line 720 didn't jump to line 721 because the condition on line 720 was never true
721 yield f"export {var.key}="
722 else:
723 yield f"export {var.key}={shlex.quote(str(val))}"
726def pretty_print_env(env: CiEnvironment, secrets: bool = False):
727 """Print environment and provider values in a human-readable format.
729 Args:
730 env: Resolved CI environment.
731 secrets: If True, include secret environment values.
732 """
733 def _render_value(value) -> str:
734 if isinstance(value, str):
735 return f'"{value}"'
736 if value is None: 736 ↛ 737line 736 didn't jump to line 737 because the condition on line 736 was never true
737 return ""
738 return str(value)
740 def _print_block(title: str, values: dict):
741 LOGGER.info(title)
742 LOGGER.info("-" * len(title))
744 width = max((len(key) for key in values), default=0)
745 for key in sorted(values):
746 LOGGER.info("%-*s %s", width, key, _render_value(values[key]))
748 LOGGER.info("")
750 LOGGER.info("Provider: %s", env.provider)
751 LOGGER.info("")
753 _print_block("Environment", env.as_dict(secrets=secrets))
755 provider_values = {
756 "pipeline_id": env.pipeline_id,
757 "pipeline_alias": env.pipeline_alias,
758 "branch_name": env.branch_name,
759 "commit_sha": env.commit_sha,
760 "commit_short_sha": env.commit_short_sha,
761 "commit_title": env.commit_title,
762 "commit_message": env.commit_message,
763 "commit_author": env.commit_author,
764 "commit_timestamp": env.commit_timestamp,
765 }
767 _print_block("Provider Metadata", provider_values)