Coverage for /builds/svesch/cikit/src/cikit/data/client.py: 100%

1# SPDX-License-Identifier: GPL-3.0-or-later 

2"""Low-level HTTP client for the cikit data API. 

3 

4This module provides the `CikitDataClient` class for direct access to the 

5remote cikit data API endpoints. It supports creating pipeline entries, 

6updating run data, querying timeline data, and fetching latest values. 

7 

8Author: 

9 MS CoPilot / OpenAI 

10Date: 

11 2026-02-12 

12""" 

13 

14from enum import StrEnum 

15 

16import requests 

17from requests.adapters import HTTPAdapter 

18from urllib3.util.retry import Retry 

19 

20from cikit.errors import CiToolsError, CikitSyncError 

21 

22 

23REQUESTS_CONNECT_TIMEOUT = 5 

24REQUESTS_READ_TIMEOUT = 30 

25REQUESTS_TIMEOUT = (REQUESTS_CONNECT_TIMEOUT, REQUESTS_READ_TIMEOUT) 

26REQUESTS_RETRY_COUNT = 3 

27REQUESTS_RETRY_BACKOFF = 1.0 

28REQUESTS_RETRY_STATUS_CODES = [500, 502, 503, 504] 

29 

30 

31def create_session() -> requests.Session: 

32 """Create a configured HTTP session. 

33 

34 The session retries temporary server-side failures and connection issues. 

35 Client-side errors are not retried because they usually indicate invalid 

36 configuration or invalid request data. 

37 

38 Returns: 

39 Configured requests session. 

40 """ 

41 retry = Retry(total=REQUESTS_RETRY_COUNT, connect=REQUESTS_RETRY_COUNT, 

42 read=REQUESTS_RETRY_COUNT, backoff_factor=REQUESTS_RETRY_BACKOFF, 

43 status_forcelist=REQUESTS_RETRY_STATUS_CODES, allowed_methods=["GET", "POST"]) 

44 

45 adapter = HTTPAdapter(max_retries=retry) 

46 session = requests.Session() 

47 session.mount("http://", adapter) 

48 session.mount("https://", adapter) 

49 

50 return session 

51 

52 

53class RefKey(StrEnum): 

54 """Supported reference keys for timeline queries.""" 

55 ID = "id" 

56 INDEX_DESC = "index_desc" 

57 TIMESTAMP = "timestamp" 

58 COMMIT_HASH = "commit_hash" 

59 SHORT_COMMIT_HASH = "short_commit_hash" 

60 PIPELINE_ID = "pipeline_id" 

61 

62 

63class CikitDataClient: 

64 """Low-level client for the remote cikit data API. 

65 

66 The client wraps the available HTTP endpoints and returns parsed JSON 

67 responses as Python dictionaries. 

68 """ 

69 def __init__(self, base_url: str, token: str = None): 

70 """Initialize the CikitDataClient. 

71 

72 Args: 

73 base_url: Base URL of the cikit data API. Must not be empty. 

74 token: Optional authentication token. If set, an ``X-Auth-Token`` 

75 header is added to all requests. 

76 

77 Raises: 

78 CiToolsError: If ``base_url`` is empty or None. 

79 """ 

80 if not base_url: 

81 raise CiToolsError("base_url must be provided explicitly") 

82 

83 self.base_url = base_url.rstrip("/") 

84 self.token = token 

85 self.session = create_session() 

86 

87 self.headers = {"Content-Type": "application/json"} 

88 if self.token: 

89 self.headers["X-Auth-Token"] = self.token 

90 

91 def _safe_request(self, method: str, url: str, **kwargs) -> dict: 

92 """Perform an HTTP request and return the parsed JSON response. 

93 

94 All request-related errors are converted to ``CikitSyncError``. 

95 

96 Args: 

97 method: HTTP method, for example ``"GET"`` or ``"POST"``. 

98 url: Fully qualified API endpoint URL. 

99 **kwargs: Additional keyword arguments forwarded to 

100 ``requests.Session.request()``, such as ``json``, ``params``, 

101 ``headers``, or ``timeout``. 

102 

103 Returns: 

104 Parsed JSON response from the API. 

105 

106 Raises: 

107 CikitSyncError: If the request fails, the response status is not 

108 successful, or the response body is not valid JSON. 

109 """ 

110 try: 

111 timeout = kwargs.pop("timeout", REQUESTS_TIMEOUT) 

112 response = self.session.request(method, url, timeout=timeout, **kwargs) 

113 response.raise_for_status() 

114 except requests.RequestException as exp: 

115 raise CikitSyncError(f"API request failed: {exp}") from exp 

116 

117 try: 

118 data = response.json() 

119 except ValueError as exp: 

120 raise CikitSyncError(f"Error while parsing API response: {exp}") from exp 

121 

122 return data 

123 

124 def create_pipeline(self, project: str, pipeline_id: str, ci_provider: str, 

125 commit_hash: str = None) -> dict: 

126 """Create a new pipeline entry. 

127 

128 Args: 

129 project: Project name. 

130 pipeline_id: External pipeline identifier. 

131 ci_provider: CI provider name. 

132 commit_hash: Optional commit hash. 

133 

134 Returns: 

135 Parsed JSON response from the API. 

136 """ 

137 url = f"{self.base_url}/cikit_data_create.php" 

138 payload = {"project": project, "pipeline_id": pipeline_id, "ci_provider": ci_provider} 

139 if commit_hash: 

140 payload["commit_hash"] = commit_hash 

141 return self._safe_request("POST", url, json=payload, headers=self.headers) 

142 

143 def update_run(self, project: str, pipeline_id: str, run: str, data: dict) -> dict: 

144 """Update run-level data for a pipeline. 

145 

146 Args: 

147 project: Project name. 

148 pipeline_id: Pipeline identifier. 

149 run: Cikit run name. 

150 data: Metric key/value pairs for this run. 

151 

152 Returns: 

153 Parsed JSON response from the API. 

154 """ 

155 url = f"{self.base_url}/cikit_data_update.php" 

156 payload = {"project": project, "pipeline_id": pipeline_id, "run": run, "data": data} 

157 return self._safe_request("POST", url, json=payload, headers=self.headers) 

158 

159 def get_timeline(self, project: str, keys: list[str], 

160 ref_key: RefKey = RefKey.TIMESTAMP, limit: int = None) -> dict: 

161 """Retrieve timeline data for one or more cikit data keys. 

162 

163 Args: 

164 project: Project name. 

165 keys: Fully qualified cikit data keys, for example 

166 ``"runs.build.duration"``. 

167 ref_key: Reference key used for the X-axis. 

168 limit: Optional maximum number of entries. 

169 

170 Returns: 

171 Parsed JSON response containing timeline references and series data. 

172 """ 

173 url = f"{self.base_url}/cikit_data_timeline.php" 

174 params = {"project": project, "ref_key": ref_key, "keys": ",".join(keys)} 

175 

176 if limit is not None: 

177 params["limit"] = str(limit) 

178 

179 return self._safe_request("GET", url, params=params, headers=self.headers) 

180 

181 def get_latest(self, project: str, key: str) -> dict: 

182 """Retrieve the latest value for a given project key. 

183 

184 Args: 

185 project: Project name. 

186 key: Fully qualified cikit data key, for example 

187 ``"runs.build.duration"``. 

188 

189 Returns: 

190 Parsed JSON response containing timestamp, pipeline_id, 

191 ci_provider, commit_hash, and value. 

192 """ 

193 url = f"{self.base_url}/cikit_data_latest.php" 

194 params = {"project": project, "key": key} 

195 return self._safe_request("GET", url, params=params, headers=self.headers)