Coverage for /builds/svesch/cikit/src/cikit/data/cli.py: 86%
64 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-09 14:17 +0000
1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-3.0-or-later
3"""Command-line interface for the cikit data Python package.
5This script provides access to the high-level cikit data API, including
6pipeline creation, run data updates, latest-value lookup, and timeline queries.
8Author:
9 Sven Schlender
10Date:
11 2026-09-20
12"""
14import argparse
15import json
16import logging
17import sys
18from pathlib import Path
19from typing import Any
21from cikit.errors import CiToolsError
22from cikit.env import CiEnvironment, get_env_vars
23from cikit.factory import build_cikit_data_from_env
25from .client import RefKey
28def _parse_data(pairs: list[str]) -> dict[str, Any]:
29 """Convert a list of key=value strings into a data dictionary.
31 Each element must contain exactly one '=' separating the key and value.
32 Values are automatically converted to int or float when possible;
33 otherwise they remain strings.
35 Args:
36 pairs: List of "key=value" formatted strings.
38 Returns:
39 Dictionary mapping data names to parsed values.
41 Raises:
42 ValueError: If any element does not contain an '=' separator.
43 """
44 result = {}
46 for pair in pairs:
47 if "=" not in pair:
48 raise ValueError(f"Invalid data format: {pair}")
50 key, value = pair.split("=", 1)
52 try:
53 value = int(value)
54 except ValueError:
55 try:
56 value = float(value)
57 except ValueError:
58 pass
60 result[key] = value
62 return result
65def _configure_logging(verbosity: int) -> None:
66 """Configure logging based on verbosity level.
68 Args:
69 verbosity: Verbosity level from CLI.
70 """
71 if verbosity <= 0: 71 ↛ 75line 71 didn't jump to line 75 because the condition on line 71 was always true
72 logging.disable(logging.CRITICAL)
73 return
75 if verbosity == 1:
76 level = logging.INFO
77 else:
78 level = logging.DEBUG
80 logging.basicConfig(level=level, format="%(levelname)s %(message)s")
83def _parse_args() -> argparse.Namespace: # pragma: no cover
84 """Build and return the parsed arguments of the cikit data CLI.
86 This function defines all subcommands and their respective arguments. It is
87 separated from ``main()`` to improve testability and modularity.
89 Returns:
90 Parsed command line arguments.
91 """
92 parser = argparse.ArgumentParser(description="CLI for cikit data")
94 parser.add_argument("-v", "--verbose", action="count", default=0,
95 help="Increase verbosity. Can be used multiple times.")
96 parser.add_argument("--project", required=False,
97 help="The name of the project to work on.")
98 parser.add_argument("--cache-dir", type=Path, required=False, default=None,
99 help="Local cache directory. If set, local caching is enabled.")
101 sub = parser.add_subparsers(dest="command", required=True)
103 sub.add_parser("create", help="Create a new pipeline entry")
105 update = sub.add_parser("update", help="Update data for a run")
106 update.add_argument("--run", required=True, help="Run name to update.")
107 update.add_argument("--data", action="append", default=[],
108 help="Data key=value pair. Can be repeated.")
110 timeline = sub.add_parser("timeline", help="Get timeline data")
111 timeline.add_argument("--run", required=True, help="Run name to query.")
112 timeline.add_argument("--key", action="append", required=True,
113 help="Key within the run data. Repeat for multiple keys.")
114 timeline.add_argument("--ref-key", default=RefKey.TIMESTAMP, choices=list(RefKey),
115 help="Reference key for the X-axis.")
116 timeline.add_argument("--limit", type=int, required=False, default=None,
117 help="Optional maximum number of timeline entries.")
119 latest = sub.add_parser("latest", help="Get the newest value for a key")
120 latest.add_argument("--run", required=True, help="Run name to query.")
121 latest.add_argument("--key", required=True, help="Key within the run data.")
123 return parser.parse_args()
126def _main_unsafe(args: argparse.Namespace) -> None:
127 """Dispatch the selected CLI command.
129 Builds the CikitData instance from the current environment and executes
130 the requested subcommand. The resulting data is written as JSON to stdout.
132 Args:
133 args: Parsed command line arguments.
135 Raises:
136 CiToolsError: If command execution fails.
137 """
138 env = CiEnvironment(get_env_vars())
139 pipeline_id, cikit_data = build_cikit_data_from_env(env)
141 if args.project: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 cikit_data.project = args.project
144 if args.cache_dir: 144 ↛ 147line 144 didn't jump to line 147 because the condition on line 144 was always true
145 cikit_data.cache_dir = args.cache_dir
147 if args.command == "create":
148 ci_provider = str(env.provider)
149 commit_hash = env.commit_sha
150 logging.info("Creating new pipeline entry with id '%s' for project '%s'!",
151 pipeline_id, cikit_data.project)
152 result = cikit_data.create(pipeline_id, ci_provider, commit_hash)
153 elif args.command == "update":
154 new_data = _parse_data(args.data)
155 result = cikit_data.update_run(pipeline_id, args.run, new_data)
156 elif args.command == "timeline":
157 result = cikit_data.timeline(run=args.run, keys=args.key, ref_key=RefKey(args.ref_key),
158 limit=args.limit)
159 elif args.command == "latest": 159 ↛ 162line 159 didn't jump to line 162 because the condition on line 159 was always true
160 result = cikit_data.latest(run=args.run, key=args.key)
161 else:
162 raise CiToolsError(f"Unknown command: {args.command}")
164 print(json.dumps(result, indent=2))
167def main():
168 """Entry point for the cikit data CLI."""
169 args = _parse_args()
170 _configure_logging(args.verbose)
172 try:
173 _main_unsafe(args)
174 except CiToolsError as exp:
175 print(f"Error: {exp}", file=sys.stderr)
176 sys.exit(1)
179if __name__ == "__main__":
180 main()