Coverage for /builds/svesch/cikit/src/cikit/data/api.py: 95%
89 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"""High-level Python API for the cikit data package.
4This module provides access to locally cached and remotely stored cikit data.
5It supports pipeline creation, run data updates, timeline queries, and
6latest-value lookup.
8Author:
9 Sven Schlender
10Date:
11 2026-02-13
12"""
14import logging
15from pathlib import Path
16from typing import Any
18from cikit.errors import CiRecoverableError, CiToolsError
20from .client import CikitDataClient, RefKey
21from .common import (get_run_filename, load_run_data, lookup_key, save_run_data)
23LOGGER = logging.getLogger(__name__)
26class CikitData:
27 """High-level API for local and remote cikit data access.
29 The class combines optional local cache handling with optional remote
30 API access through ``CikitDataClient``. Cached values are preferred where
31 applicable, while remote access is used for operations that require
32 server-side data.
33 """
35 def __init__(self, project: str, cache_dir: Path = None, client: CikitDataClient = None):
36 """Initialize a CikitData instance.
38 At least one data source must be configured: a local cache directory,
39 a remote ``CikitDataClient``, or both.
41 Args:
42 project: Name of the project.
43 cache_dir: Directory containing cached run data files.
44 client: Optional remote API client.
46 Raises:
47 CiToolsError: If neither ``cache_dir`` nor ``client`` is provided.
48 """
49 if not cache_dir and not client:
50 raise CiToolsError("Either 'cache_dir' or 'client' must be provided.")
52 self.project = project
53 self.cache_dir = cache_dir
54 self.client = client
56 def create(self, pipeline_id: str, ci_provider: str, commit_hash: str = None) -> dict:
57 """Create a new pipeline entry via the remote API.
59 If no remote client is configured, an empty dict is returned. This
60 allows cache-only usage without raising an error.
62 Args:
63 pipeline_id: Pipeline identifier.
64 ci_provider: Name of the CI provider.
65 commit_hash: Optional commit hash.
67 Returns:
68 Parsed JSON response from the API, or an empty dict if no remote
69 client is configured.
70 """
71 if self.client is None:
72 return {}
74 return self.client.create_pipeline(self.project, pipeline_id, ci_provider, commit_hash)
76 def update_run(self, pipeline_id: str, run: str, data: dict) -> dict:
77 """Update data for a logical run.
79 The provided values are merged into the local cache if ``cache_dir`` is
80 configured. If a remote client is available, only the newly provided
81 ``data`` is sent to the server.
83 Args:
84 pipeline_id: Pipeline identifier for remote updates.
85 run: Logical run name.
86 data: New key/value pairs to merge into the run data.
88 Returns:
89 Parsed JSON response from the API if a remote client is configured.
90 Otherwise, the merged local run data.
92 Raises:
93 CiToolsError: If remote access is used without a pipeline id.
94 """
95 cache_file = None
96 if self.cache_dir: 96 ↛ 99line 96 didn't jump to line 99 because the condition on line 96 was always true
97 cache_file = get_run_filename(self.cache_dir, run)
99 merged_data = data
100 if cache_file is not None: 100 ↛ 105line 100 didn't jump to line 105 because the condition on line 100 was always true
101 old_data = load_run_data(cache_file)
102 merged_data = {**old_data, **data}
103 save_run_data(cache_file, merged_data)
105 if self.client is None:
106 return merged_data
108 if pipeline_id is None:
109 raise CiToolsError("Remote update requires pipeline-id!")
111 return self.client.update_run(self.project, pipeline_id, run, data)
113 def timeline(self, run: str, keys: list, ref_key: RefKey = RefKey.TIMESTAMP,
114 limit: int = None) -> dict:
115 """Return timeline data for one or more run keys.
117 Timeline queries always use the remote API. Local cache data is not used
118 for this operation.
120 Args:
121 run: Logical run name.
122 keys: Data keys to query inside the selected run.
123 ref_key: Reference key used for the X-axis.
124 limit: Optional maximum number of timeline entries returned by the API.
126 Returns:
127 Dictionary with timeline data in the form:
128 {
129 "refs": [...],
130 "series": {
131 "<key>": [...]
132 }
133 }
134 The keys in ``series`` correspond to the requested run data keys.
136 Raises:
137 CiRecoverableError: If no remote client is configured.
138 """
139 if self.client is None:
140 raise CiRecoverableError("The 'timeline' command requires remote access.")
142 prefix = f"runs.{run}."
143 remote_keys = [f"{prefix}{key}" for key in keys]
144 data = self.client.get_timeline(project=self.project, keys=remote_keys, ref_key=ref_key,
145 limit=limit)
147 refs = data.get("refs", [])
148 series = {}
150 for entry in data.get("series", []):
151 key = entry.get("key")
152 values = entry.get("values")
154 if key is None or values is None: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 continue
157 if key.startswith(prefix): 157 ↛ 160line 157 didn't jump to line 160 because the condition on line 157 was always true
158 key = key[len(prefix):]
160 series[key] = values
162 return {"refs": refs, "series": series}
164 def latest(self, run: str, key: str) -> Any:
165 """Return the latest value for a single run key.
167 The local cache is checked first. If the value is not available there,
168 the remote API is queried. Only the resolved value is returned, not the
169 full API response.
171 Args:
172 run: Logical run name.
173 key: Key inside the data of the selected run.
175 Returns:
176 Resolved value from the local cache or remote API.
178 Raises:
179 CiToolsError: If neither cache nor remote access can provide the
180 value, or if the remote response is invalid.
181 """
182 if self.cache_dir: 182 ↛ 190line 182 didn't jump to line 190 because the condition on line 182 was always true
183 cache_file = get_run_filename(self.cache_dir, run)
184 cache_data = load_run_data(cache_file)
186 value = lookup_key(cache_data, key)
187 if value is not None:
188 return value
190 if self.client is not None:
191 remote_key = f"runs.{run}.{key}"
192 remote_response = self.client.get_latest(project=self.project, key=remote_key)
194 if not isinstance(remote_response, dict) or "value" not in remote_response:
195 raise CiToolsError(f"Remote response for key '{remote_key}' "
196 "does not contain a 'value' field.")
198 return remote_response["value"]
200 raise CiToolsError("The 'latest' command requires either cached data or remote access.")
203def build_cikit_data(project: str, cache_dir: Path = None,
204 base_url: str = None, token: str = None) -> CikitData:
205 """Create and return a configured CikitData instance.
207 If ``base_url`` is provided, a ``CikitDataClient`` is created
208 automatically. Otherwise, the returned instance operates in cache-only
209 mode.
211 Args:
212 project: Name of the project.
213 cache_dir: Directory for cached run data files.
214 base_url: Base URL of the remote cikit data API.
215 token: Optional authentication token for the remote API.
217 Returns:
218 Configured ``CikitData`` instance.
220 Raises:
221 CiToolsError: If neither ``cache_dir`` nor ``base_url`` is provided.
222 """
223 if not cache_dir and not base_url:
224 raise CiToolsError("Either 'cache_dir' or 'base_url' must be provided.")
226 client = None
227 if base_url:
228 client = CikitDataClient(base_url, token)
230 return CikitData(project, cache_dir, client)
233class RunContext:
234 """Run-bound convenience wrapper around ``CikitData``.
236 This class binds a ``CikitData`` instance to a fixed run name and
237 pipeline id, simplifying repeated operations on the same run.
238 """
240 def __init__(self, cikit_data: CikitData, pipeline_id: str, run_name: str):
241 """Initialize a RunContext.
243 Args:
244 cikit_data: Shared ``CikitData`` instance.
245 pipeline_id: Pipeline identifier used for remote updates.
246 run_name: Logical run name bound to this context.
247 """
248 self._data = cikit_data
249 self._pipeline_id = pipeline_id
250 self.run_name = run_name
252 def set(self, key: str, value) -> dict:
253 """Set a single data value for the bound run.
255 Args:
256 key: Data key to set.
257 value: Value to assign.
259 Returns:
260 Parsed JSON response from the API if remote is used, otherwise
261 the merged local run data.
262 """
263 return self._data.update_run(self._pipeline_id, self.run_name, {key: value})
265 def update(self, data: dict) -> dict:
266 """Merge multiple data values for the bound run.
268 Args:
269 data: Dictionary of key/value pairs to merge.
271 Returns:
272 Parsed JSON response from the API if remote is used, otherwise
273 the merged local run data.
274 """
275 return self._data.update_run(self._pipeline_id, self.run_name, data)
277 def latest(self, key: str):
278 """Return the latest value for a key of the bound run.
280 Args:
281 key: Data key to resolve.
283 Returns:
284 Resolved value from cache or remote API.
286 Raises:
287 CiToolsError: If the value cannot be resolved.
288 """
289 return self._data.latest(self.run_name, key)
291 def timeline(self, keys: list, ref_key: RefKey = RefKey.TIMESTAMP,
292 limit: int = None) -> dict:
293 """Return timeline data for one or more keys of the bound run.
295 Args:
296 keys: Data keys to query.
297 ref_key: Reference key used for the X-axis.
298 limit: Optional maximum number of timeline entries returned by the API.
300 Returns:
301 Timeline data containing ``refs`` and ``series``.
303 Raises:
304 CiRecoverableError: If no remote client is configured.
305 """
306 return self._data.timeline(self.run_name, keys, ref_key, limit)
308 def get_data(self) -> dict:
309 """Return the current run data as a dictionary copy.
311 The data is loaded from the local cache if available. Remote data is
312 not queried.
314 Returns:
315 A copy of the current run data dictionary.
317 Raises:
318 CiRecoverableError: If no local cache directory is configured.
319 """
320 if self._data.cache_dir is None:
321 raise CiRecoverableError("get_data() requires a local cache_dir.")
323 filepath = get_run_filename(self._data.cache_dir, self.run_name)
324 data = load_run_data(filepath)
326 return dict(data) if isinstance(data, dict) else {}