Coverage for pyTooling/Tracing/CI/GitHub.py: 94%
181 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-03 23:02 +0000
1# ==================================================================================================================== #
2# _____ _ _ _____ _ #
3# _ __ _ |_ _|__ ___ | (_)_ __ __ _|_ _| __ __ _ ___(_)_ __ __ _ #
4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` | | || '__/ _` |/ __| | '_ \ / _` | #
5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_| || | | (_| | (__| | | | | (_| | #
6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_||_| \__,_|\___|_|_| |_|\__, | #
7# |_| |___/ |___/ |___/ #
8# ==================================================================================================================== #
9# Authors: #
10# Patrick Lehmann #
11# #
12# License: #
13# ==================================================================================================================== #
14# Copyright 2026-2026 Patrick Lehmann - Bötzingen, Germany #
15# #
16# Licensed under the Apache License, Version 2.0 (the "License"); #
17# you may not use this file except in compliance with the License. #
18# You may obtain a copy of the License at #
19# #
20# http://www.apache.org/licenses/LICENSE-2.0 #
21# #
22# Unless required by applicable law or agreed to in writing, software #
23# distributed under the License is distributed on an "AS IS" BASIS, #
24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
25# See the License for the specific language governing permissions and #
26# limitations under the License. #
27# #
28# SPDX-License-Identifier: Apache-2.0 #
29# ==================================================================================================================== #
30#
31"""
32Read the timing of a GitHub Actions workflow run into a software execution trace.
34A workflow run becomes a :class:`~pyTooling.Tracing.Trace`, and each job a :class:`~pyTooling.Tracing.Span` with one
35sub-span per step. The time a job waited for a runner is a separate timespan in front of the job. Jobs of a called
36(reusable) workflow - which GitHub names ``Caller / Job`` - are grouped below a timespan named like the calling job.
38.. code-block:: python
40 from os import getenv
41 from pyTooling.Tracing.CI.GitHub import WorkflowRunReader
43 reader = WorkflowRunReader("pyTooling/Actions", token=getenv("GITHUB_TOKEN"))
44 trace = reader.ReadRun(34937615362)
45 print("\\n".join(trace.Format()))
47.. hint::
49 See :ref:`high-level help <TRACING/CI>` for explanations and usage examples.
50"""
51from datetime import datetime
52from typing import Any, ClassVar, Iterable, Optional as Nullable, Self, Union
54from pyTooling.CI import Base, JobGroup, JSONObject, Matrix
55from pyTooling.CI.GitHub import Conclusion, GitHubError, Job, MatrixJob, Pipeline, Step
56from pyTooling.Common import getFullyQualifiedName
57from pyTooling.Decorators import export, readonly
58from pyTooling.GenericPath.URL import URL
59from pyTooling.MetaClasses import ExtendedType
60from pyTooling.REST import RESTClient, RESTError
61from pyTooling.Tracing import Span, Trace
62from pyTooling.Tracing.CI import Result
63from pyTooling.Tracing.CI import JobSpan as CIJobSpan, MatrixSpan as CIMatrixSpan
64from pyTooling.Tracing.CI import PipelineTrace as CIPipelineTrace, QueuedSpan as CIQueuedSpan
65from pyTooling.Tracing.CI import StepSpan as CIStepSpan, WorkflowSpan as CIWorkflowSpan
68__all__ = ["GITHUB_API_URL"]
70GITHUB_API_URL = URL.Parse("https://api.github.com")
71"""Base URL of the GitHub REST API."""
73_GITHUB_HEADERS = {
74 "Accept": "application/vnd.github+json",
75 "X-GitHub-Api-Version": "2022-11-28"
76}
77"""Headers every request to the GitHub REST API carries."""
80@export
81class GitHub(metaclass=ExtendedType, slots=True):
82 """
83 Attribute keys naming what only GitHub Actions reports, beside the keys of
84 :class:`~pyTooling.Tracing.CI.OTLP`.
86 The nesting mirrors the key itself, as it does there: :attr:`GitHub.Runner.Labels` spells
87 ``'github.runner.labels'``.
88 """
90 Conclusion: ClassVar[str] = "github.conclusion" #: GitHub's own conclusion, next to the result it maps to.
91 Event: ClassVar[str] = "github.event" #: The event that started the run, e.g. ``'push'``.
93 class Run(metaclass=ExtendedType, slots=True):
94 """Attribute keys naming a workflow run."""
96 Attempt: ClassVar[str] = "github.run.attempt" #: Which attempt of the run this is.
97 Number: ClassVar[str] = "github.run.number" #: The run's number within its workflow.
99 class Workflow(metaclass=ExtendedType, slots=True):
100 """Attribute keys naming the workflow a run belongs to."""
102 Path: ClassVar[str] = "github.workflow.path" #: The workflow's YAML file in the repository.
104 class Matrix(metaclass=ExtendedType, slots=True):
105 """Attribute keys naming a matrix."""
107 Dimensions: ClassVar[str] = "github.matrix.dimensions" #: One instance's values, e.g. ``['ubuntu-26.04', '3.14']``.
109 class Runner(metaclass=ExtendedType, slots=True):
110 """Attribute keys naming the runner a job ran on."""
112 Labels: ClassVar[str] = "github.runner.labels" #: The labels the job asked for, e.g. ``['ubuntu-26.04']``.
113 Group: ClassVar[str] = "github.runner.group" #: The runner group the runner belongs to.
115 class Step(metaclass=ExtendedType, slots=True):
116 """Attribute keys naming a step of a job."""
118 Number: ClassVar[str] = "github.step.number" #: The step's position in its job, counted from one.
121@export
122class GitHubTimespanMixin(metaclass=ExtendedType, mixin=True):
123 """
124 Mixin-class for a timespan built from :mod:`pyTooling.CI.GitHub`'s model of a workflow run.
126 It holds what every flavour needs to read that model: the result an element's outcome is, and the span a group's
127 contents have. Everything else the model answers itself - the order its elements are in, and the times of a job, which
128 are wider than the ones GitHub reports because a step may run outside the job containing it.
129 """
131 @staticmethod
132 def _Result(element: Base) -> Nullable[Result]:
133 """
134 Return the CI/CD result of an element of the model.
136 The result is the member of the same value as the element's :attr:`~pyTooling.CI.Base.Outcome`. A
137 GitHub conclusion is mapped onto an outcome by
138 :meth:`Conclusion.ToOutcome <pyTooling.CI.GitHub.Conclusion.ToOutcome>`.
140 :param element: The workflow run, job or step.
141 :returns: The CI/CD result, or ``None`` while the element hasn't ended.
142 """
143 if element.Outcome is None:
144 return None
146 return Result(element.Outcome.value)
148 @staticmethod
149 def _NotBefore(end: Nullable[datetime], begin: datetime) -> Nullable[datetime]:
150 """
151 Clamp an end time so it doesn't precede a begin time.
153 :param end: Optional, the end time. Default: ``None``.
154 :param begin: The begin time.
155 :returns: The end time, but not before the begin time.
156 """
157 return None if end is None else max(end, begin)
159 @classmethod
160 def _Timespan(
161 cls,
162 created: Nullable[datetime],
163 started: Nullable[datetime],
164 completed: Nullable[datetime]
165 ) -> tuple[Nullable[datetime], Nullable[datetime]]:
166 """
167 Return the timespan the model's three times describe.
169 The model spans what a group holds; this is where the span becomes a timespan: it begins when the first
170 element was queued, or started if it was never queued, and it doesn't end before it begins.
172 :param created: When the first element was created, or ``None`` if none reports a time.
173 :param started: When the first element started, or ``None`` if none has started.
174 :param completed: When the last element completed, or ``None`` while one hasn't.
175 :returns: When the timespan begins and ends, each ``None`` if there is nothing to place on a
176 timeline respectively while it is still running.
177 """
178 begin = created if created is not None else started
179 if begin is None:
180 return None, None
182 return begin, cls._NotBefore(completed, begin)
184 @classmethod
185 def _AddContents(cls, group: JobGroup, parent: Span) -> None:
186 """
187 Add the jobs, matrices and called workflows of a group to a timespan, ordered by the time they were queued.
189 :param group: The group - a workflow run, a called workflow or a matrix.
190 :param parent: The timespan the group's contents are added to.
191 """
192 for item in group.IterateElements():
193 if isinstance(item, Job):
194 JobSpan.FromJob(item, parent)
195 elif isinstance(item, Matrix):
196 MatrixSpan.FromGroup(item, parent)
197 else:
198 WorkflowSpan.FromGroup(item, parent)
201@export
202class StepSpan(CIStepSpan, GitHubTimespanMixin):
203 """The timespan of a step of a GitHub Actions job."""
205 @classmethod
206 def FromStep(cls, step: Step, parent: Span) -> Nullable[Self]:
207 """
208 Build the timespan of a step, below the timespan of its job.
210 A step that never started has no timespan - there is nothing to place on a timeline.
212 :param step: The step.
213 :param parent: The timespan of the job containing the step.
214 :returns: The step's timespan, or ``None`` if the step never started.
215 """
216 if step.StartedAt is None:
217 return None
219 return cls(
220 step.Name,
221 step.StartedAt,
222 cls._NotBefore(step.CompletedAt, step.StartedAt),
223 parent=parent,
224 result=cls._Result(step),
225 attributes={
226 GitHub.Step.Number: step.Number,
227 GitHub.Conclusion: None if step.Conclusion is None else step.Conclusion.value
228 }
229 )
232@export
233class QueuedSpan(CIQueuedSpan, GitHubTimespanMixin):
234 """The timespan a GitHub Actions job waited for a runner, in front of the job's own timespan."""
236 @classmethod
237 def FromJob(cls, job: Job, created: datetime, started: Nullable[datetime], parent: Span) -> Self:
238 """
239 Build the timespan a job waited for a runner.
241 :param job: The job that waited.
242 :param created: When the job was created, which is when it started waiting.
243 :param started: Optional, when the job started, or ``None`` while it is still waiting.
244 :param parent: The timespan of the trace, workflow or matrix containing the job.
245 :returns: The waiting timespan.
246 """
247 return cls(
248 f"{job!s} (queued)",
249 created,
250 cls._NotBefore(started, created),
251 parent=parent,
252 taskName=job.QualifiedName,
253 attributes={GitHub.Runner.Labels: list(job.Labels)}
254 )
257@export
258class JobSpan(CIJobSpan, GitHubTimespanMixin):
259 """The timespan of a GitHub Actions job, from the moment it started on a runner until it completed."""
261 @classmethod
262 def FromJob(cls, job: Job, parent: Span) -> Nullable[Self]:
263 """
264 Build the timespans of a job: the time it waited for a runner, the job itself, and one per step.
266 A step may be reported as starting before, or completing after, the job containing it, so the job's timespan is
267 widened to hold its steps - :mod:`pyTooling.Tracing` requires a timespan to lie within its parent.
269 :param job: The job.
270 :param parent: The timespan of the trace, workflow or matrix containing the job.
271 :returns: The job's timespan, or ``None`` if the job hasn't started and therefore only waited.
272 """
273 created, started, completed = job.CreatedAt, job.StartedAt, job.CompletedAt
275 # a matrix instance carries its dimension values, so two of them are distinguishable
276 displayName = str(job)
277 if job.Conclusion is Conclusion.Skipped:
278 # A skipped job neither waited for a runner nor ran on one.
279 begin = started if started is not None else created
280 end = None if begin is None else cls._NotBefore(completed if completed is not None else begin, begin)
281 else:
282 if created is not None and (started is None or created < started):
283 QueuedSpan.FromJob(job, created, started, parent)
285 if started is None:
286 return None
288 begin = started
289 end = cls._NotBefore(completed, started)
291 attributes = {
292 GitHub.Conclusion: None if job.Conclusion is None else job.Conclusion.value,
293 GitHub.Runner.Group: job.RunnerGroupName,
294 GitHub.Runner.Labels: list(job.Labels)
295 }
296 if isinstance(job, MatrixJob):
297 attributes[GitHub.Matrix.Dimensions] = list(job.Dimensions.values())
299 jobSpan = cls(
300 displayName,
301 begin,
302 end,
303 parent=parent,
304 taskName=job.QualifiedName,
305 runID=None if job.ID is None else str(job.ID),
306 runURL=None if job.URL is None else str(job.URL),
307 result=cls._Result(job),
308 workerName=job.RunnerName,
309 attributes=attributes
310 )
312 for step in job.Steps:
313 StepSpan.FromStep(step, jobSpan)
315 return jobSpan
318@export
319class GitHubGroupMixin(metaclass=ExtendedType, mixin=True):
320 """
321 Mixin-class for a timespan holding other timespans - a called workflow or a matrix.
323 Both are built the same way and differ only in what they are, which their class says. Its
324 :meth:`~GitHubTimespanMixin._Timespan` and :meth:`~GitHubTimespanMixin._AddContents` come from
325 :class:`GitHubTimespanMixin`.
326 """
328 @classmethod
329 def FromGroup(cls, group: JobGroup, parent: Span) -> Self:
330 """
331 Build the timespan of a group and everything below it.
333 :param group: The called workflow or the matrix.
334 :param parent: The timespan containing the group.
335 :returns: The group's timespan.
336 """
337 begin, end = cls._Timespan(group.CreatedAt, group.StartedAt, group.CompletedAt)
338 span = cls(str(group), begin, end, parent=parent)
339 cls._AddContents(group, span)
341 return span
344@export
345class WorkflowSpan(CIWorkflowSpan, GitHubTimespanMixin, GitHubGroupMixin):
346 """The jobs of a called workflow, grouped into one timespan."""
349@export
350class MatrixSpan(CIMatrixSpan, GitHubTimespanMixin, GitHubGroupMixin):
351 """The instances a matrix produced, grouped into one timespan."""
354@export
355class WorkflowRunTrace(CIPipelineTrace, GitHubTimespanMixin):
356 """A GitHub Actions workflow run as a trace."""
358 @classmethod
359 def FromJSON(cls, run: JSONObject, jobs: Nullable[Iterable[JSONObject]] = None) -> Self:
360 """
361 Build the trace of a workflow run from the payloads the GitHub REST API answers with.
363 The payloads are read into a :class:`~pyTooling.CI.GitHub.Pipeline` first, so the tree - a called workflow, a
364 matrix, the jobs and their steps - is reconstructed by the model rather than here, and :meth:`FromPipeline`
365 turns it into timespans.
367 :param run: The workflow run, as returned by ``GET /repos/{owner}/{repo}/actions/runs/{run_id}``.
368 :param jobs: Optional, the run's jobs, as listed by ``GET .../actions/runs/{run_id}/jobs``.
369 :returns: The workflow run as a trace.
370 :raises TypeError: If parameter 'run' is not of type :class:`dict`.
371 :raises GitHubError: If a mandatory field is missing, or a field holds a value GitHub doesn't document.
372 """
373 return cls.FromPipeline(Pipeline.FromJSON(run, jobs))
375 @classmethod
376 def FromPipeline(cls, pipeline: Pipeline) -> Self:
377 """
378 Build the trace of a workflow run, as :mod:`pyTooling.CI.GitHub` models it.
380 * The run becomes the trace, widened where a job was queued earlier or completed later than the run reports.
381 * A called workflow and a matrix become a timespan holding the jobs below them.
382 * A job becomes a timespan from its start to its completion, preceded by ``<job> (queued)`` from its creation to
383 its start, if it waited. A skipped job has no waiting timespan, and a job that hasn't started yet only a running
384 waiting timespan.
385 * A step that started becomes a sub-span of its job.
386 * Every timespan is classified by :attr:`CI.Span.Kind <pyTooling.Tracing.CI.CI>` and carries the OpenTelemetry CI/CD
387 attributes, GitHub's conclusion (:data:`CONCLUSION`), and for jobs the runner's labels and group.
389 :param pipeline: The workflow run.
390 :returns: The workflow run as a trace.
391 :raises TypeError: If parameter 'pipeline' is not of type :class:`~pyTooling.CI.GitHub.Pipeline`.
392 """
393 if not isinstance(pipeline, Pipeline): 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true
394 ex = TypeError("Parameter 'pipeline' is not of type 'Pipeline'.")
395 ex.add_note(f"Got type '{getFullyQualifiedName(pipeline)}'.")
396 raise ex
398 beginTime = pipeline.StartedAt if pipeline.StartedAt is not None else pipeline.CreatedAt
399 endTime = cls._NotBefore(pipeline.CompletedAt, beginTime) if beginTime is not None else pipeline.CompletedAt
401 # a job may be queued before the run reports itself started, and may complete after the run's last update
402 jobsBegin, jobsEnd = cls._Timespan(
403 pipeline.ContentsCreatedAt, pipeline.ContentsStartedAt, pipeline.ContentsCompletedAt
404 )
405 if jobsBegin is not None:
406 beginTime = jobsBegin if beginTime is None else min(beginTime, jobsBegin)
408 if endTime is not None and jobsEnd is not None:
409 endTime = max(endTime, jobsEnd)
411 trace = cls(
412 pipeline.Name,
413 beginTime,
414 endTime,
415 runID=None if pipeline.ID is None else str(pipeline.ID),
416 runURL=None if pipeline.URL is None else str(pipeline.URL),
417 result=cls._Result(pipeline),
418 reference=pipeline.GitReference,
419 revision=pipeline.SHA,
420 attributes={
421 GitHub.Conclusion: None if pipeline.Conclusion is None else pipeline.Conclusion.value,
422 GitHub.Run.Attempt: pipeline.RunAttempt,
423 GitHub.Run.Number: pipeline.RunNumber,
424 GitHub.Workflow.Path: pipeline.Path,
425 GitHub.Event: None if pipeline.Event is None else pipeline.Event.value
426 }
427 )
428 cls._AddContents(pipeline, trace)
430 return trace
433@export
434class WorkflowRunReader(RESTClient):
435 """
436 Reads workflow runs of a GitHub repository through the GitHub REST API and converts them into traces.
438 A token is needed for a private repository, and it raises the rate limit for a public one - inside a workflow,
439 ``GITHUB_TOKEN`` with the ``actions: read`` permission suffices.
441 The requests themselves - the retries, the pagination and the token's boundary - are
442 :class:`~pyTooling.REST.RESTClient`'s.
443 """
444 _repository: str #: Repository as ``owner/name``.
446 def __init__(
447 self,
448 repository: str,
449 token: Nullable[str] = None,
450 *,
451 apiURL: Union[str, URL] = GITHUB_API_URL,
452 timeout: float = 30.0,
453 retries: int = 3,
454 retryDelay: float = 2.0
455 ) -> None:
456 """
457 Initializes a reader for the workflow runs of a repository.
459 :param repository: The repository as ``owner/name``.
460 :param token: Optional, token authorizing the requests. Default: anonymous requests.
461 :param apiURL: Optional, base URL of the GitHub REST API, e.g. of a GitHub Enterprise Server.
462 Default: :data:`GITHUB_API_URL`.
463 :param timeout: Optional, timeout of a single request in seconds. Default: ``30.0``.
464 :param retries: Optional, how often a transiently failing request is tried again. ``0`` tries once.
465 Default: ``3``.
466 :param retryDelay: Optional, pause in seconds before a request is tried again the first time. The pause doubles
467 with every further attempt. Default: ``2.0``.
468 :raises ValueError: If parameter 'repository' is ``None``.
469 :raises TypeError: If parameter 'repository' is not of type :class:`str`.
470 :raises ValueError: If parameter 'repository' isn't of the form ``owner/name``.
471 :raises TypeError: If a parameter of :class:`~pyTooling.REST.RESTClient` has the wrong type.
472 :raises ValueError: If a parameter of :class:`~pyTooling.REST.RESTClient` has an invalid value.
473 """
474 super().__init__(apiURL, token, headers=_GITHUB_HEADERS, timeout=timeout, retries=retries, retryDelay=retryDelay)
476 if repository is None:
477 raise ValueError("Parameter 'repository' is None.")
478 elif not isinstance(repository, str): 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 ex = TypeError("Parameter 'repository' is not of type 'str'.")
480 ex.add_note(f"Got type '{getFullyQualifiedName(repository)}'.")
481 raise ex
482 elif len(parts := repository.split("/")) != 2 or "" in parts:
483 ex = ValueError("Parameter 'repository' isn't of the form 'owner/name'.")
484 ex.add_note(f"Got value '{repository}'.")
485 raise ex
487 self._repository = repository
489 @readonly
490 def Repository(self) -> str:
491 """
492 Read-only property to access the repository (:attr:`_repository`).
494 :returns: The repository as ``owner/name``.
495 """
496 return self._repository
498 def ReadRun(self, runID: int, attempt: Nullable[int] = None) -> Trace:
499 """
500 Read a workflow run and all its jobs, and convert them into a trace.
502 :param runID: The workflow run's identifier.
503 :param attempt: Optional, the run attempt to read. Default: the latest attempt.
504 :returns: The workflow run as a trace (see :meth:`WorkflowRunTrace.FromJSON`).
505 :raises ValueError: If parameter 'runID' is ``None``.
506 :raises TypeError: If parameter 'runID' is not of type :class:`int`.
507 :raises ValueError: If parameter 'runID' isn't positive.
508 :raises TypeError: If parameter 'attempt' is not of type :class:`int`.
509 :raises ValueError: If parameter 'attempt' isn't positive.
510 :raises RESTError: If a request fails, or GitHub's answer isn't a JSON object.
511 :raises GitHubError: If GitHub's answer lacks a mandatory field.
512 """
513 if runID is None:
514 raise ValueError("Parameter 'runID' is None.")
515 elif isinstance(runID, bool) or not isinstance(runID, int):
516 ex = TypeError("Parameter 'runID' is not of type 'int'.")
517 ex.add_note(f"Got type '{getFullyQualifiedName(runID)}'.")
518 raise ex
519 elif runID <= 0:
520 ex = ValueError("Parameter 'runID' isn't positive.")
521 ex.add_note(f"Got value '{runID}'.")
522 raise ex
524 if attempt is not None:
525 if isinstance(attempt, bool) or not isinstance(attempt, int): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true
526 ex = TypeError("Parameter 'attempt' is not of type 'int'.")
527 ex.add_note(f"Got type '{getFullyQualifiedName(attempt)}'.")
528 raise ex
529 elif attempt <= 0:
530 ex = ValueError("Parameter 'attempt' isn't positive.")
531 ex.add_note(f"Got value '{attempt}'.")
532 raise ex
534 runPath = f"repos/{self._repository}/actions/runs/{runID}"
535 if attempt is None:
536 run, _ = self.GetJSONObject(runPath)
537 jobsPath = f"{runPath}/jobs?filter=latest&per_page=100"
538 else:
539 run, _ = self.GetJSONObject(f"{runPath}/attempts/{attempt}")
540 jobsPath = f"{runPath}/attempts/{attempt}/jobs?per_page=100"
542 jobs: list[JSONObject] = []
543 nextPath: Nullable[str] = jobsPath
544 while nextPath is not None:
545 page, nextPath = self.GetJSONObject(nextPath)
546 if not isinstance(pageJobs := page.get("jobs", None), list):
547 raise GitHubError(f"Field 'jobs' is missing in the answer of '{jobsPath}'.")
548 jobs.extend(pageJobs)
550 return WorkflowRunTrace.FromJSON(run, jobs)
552 def _AddErrorNotes(self, error: RESTError, status: int) -> None:
553 """
554 Add a note saying what a failing status means for the GitHub REST API.
556 :param error: The error the note is added to.
557 :param status: The HTTP status the request failed with.
558 """
559 if status in (401, 403, 404): 559 ↛ exitline 559 didn't return from function '_AddErrorNotes' because the condition on line 559 was always true
560 error.add_note("Check the repository's name, and that the token may read the repository's actions.")