Skip to content

Commit

Permalink
Add support for workflow jobs and steps (#1951)
Browse files Browse the repository at this point in the history
Co-authored-by: Enrico Minack <github@enrico.minack.dev>
  • Loading branch information
Tenzer and EnricoMi committed Jun 20, 2023
1 parent cb50dec commit 804c310
Show file tree
Hide file tree
Showing 11 changed files with 537 additions and 2 deletions.
201 changes: 201 additions & 0 deletions github/WorkflowJob.py
@@ -0,0 +1,201 @@
############################ Copyrights and license ############################
# #
# Copyright 2021 Jeppe Fihl-Pearson <jeppe@tenzer.dk> #
# #
# This file is part of PyGithub. #
# http://pygithub.readthedocs.io/ #
# #
# PyGithub is free software: you can redistribute it and/or modify it under #
# the terms of the GNU Lesser General Public License as published by the Free #
# Software Foundation, either version 3 of the License, or (at your option) #
# any later version. #
# #
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
# details. #
# #
# You should have received a copy of the GNU Lesser General Public License #
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################

import github.GithubObject
import github.WorkflowStep


class WorkflowJob(github.GithubObject.CompletableGithubObject):
"""
This class represents Workflow Jobs. The reference can be found here https://docs.github.com/en/rest/reference/actions#workflow-jobs
"""

def __repr__(self):
return self.get__repr__({"id": self._id.value, "url": self._url.value})

@property
def check_run_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._check_run_url)
return self._check_run_url.value

@property
def completed_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._completed_at)
return self._completed_at.value

@property
def conclusion(self):
"""
:type: string
"""
self._completeIfNotSet(self._conclusion)
return self._conclusion.value

@property
def head_sha(self):
"""
:type: string
"""
self._completeIfNotSet(self._head_sha)
return self._head_sha.value

@property
def html_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._html_url)
return self._html_url.value

@property
def id(self):
"""
:type: int
"""
self._completeIfNotSet(self._id)
return self._id.value

@property
def name(self):
"""
:type: string
"""
self._completeIfNotSet(self._name)
return self._name.value

@property
def node_id(self):
"""
:type: string
"""
self._completeIfNotSet(self._node_id)
return self._node_id.value

@property
def run_id(self):
"""
:type: integer
"""
self._completeIfNotSet(self._run_id)
return self._run_id.value

@property
def run_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._run_url)
return self._run_url.value

@property
def started_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._started_at)
return self._started_at.value

@property
def status(self):
"""
:type: string
"""
self._completeIfNotSet(self._status)
return self._status.value

@property
def steps(self):
"""
:type: list of github.WorkflowStep.WorkflowStep
"""
self._completeIfNotSet(self._steps)
return self._steps.value

@property
def url(self):
"""
:type: string
"""
self._completeIfNotSet(self._url)
return self._url.value

def logs_url(self):
"""
:type: string
"""
print(f"{self.url}/logs")
headers, _ = self._requester.requestBlobAndCheck("GET", f"{self.url}/logs")
return headers["location"]

def _initAttributes(self):
self._check_run_url = github.GithubObject.NotSet
self._completed_at = github.GithubObject.NotSet
self._conclusion = github.GithubObject.NotSet
self._head_sha = github.GithubObject.NotSet
self._html_url = github.GithubObject.NotSet
self._id = github.GithubObject.NotSet
self._name = github.GithubObject.NotSet
self._node_id = github.GithubObject.NotSet
self._run_id = github.GithubObject.NotSet
self._run_url = github.GithubObject.NotSet
self._started_at = github.GithubObject.NotSet
self._status = github.GithubObject.NotSet
self._steps = github.GithubObject.NotSet
self._url = github.GithubObject.NotSet

def _useAttributes(self, attributes):
if "check_run_url" in attributes: # pragma no branch
self._check_run_url = self._makeStringAttribute(attributes["check_run_url"])
if "completed_at" in attributes: # pragma no branch
self._completed_at = self._makeDatetimeAttribute(attributes["completed_at"])
if "conclusion" in attributes: # pragma no branch
self._conclusion = self._makeStringAttribute(attributes["conclusion"])
if "head_sha" in attributes: # pragma no branch
self._head_sha = self._makeStringAttribute(attributes["head_sha"])
if "html_url" in attributes: # pragma no branch
self._html_url = self._makeStringAttribute(attributes["html_url"])
if "id" in attributes: # pragma no branch
self._id = self._makeIntAttribute(attributes["id"])
if "name" in attributes: # pragma no branch
self._name = self._makeStringAttribute(attributes["name"])
if "node_id" in attributes: # pragma no branch
self._node_id = self._makeStringAttribute(attributes["node_id"])
if "run_id" in attributes: # pragma no branch
self._run_id = self._makeIntAttribute(attributes["run_id"])
if "run_url" in attributes: # pragma no branch
self._run_url = self._makeStringAttribute(attributes["run_url"])
if "started_at" in attributes: # pragma no branch
self._started_at = self._makeDatetimeAttribute(attributes["started_at"])
if "status" in attributes: # pragma no branch
self._status = self._makeStringAttribute(attributes["status"])
if "steps" in attributes: # pragma no branch
self._steps = self._makeListOfClassesAttribute(
github.WorkflowStep.WorkflowStep, attributes["steps"]
)
if "url" in attributes: # pragma no branch
self._url = self._makeStringAttribute(attributes["url"])
39 changes: 39 additions & 0 deletions github/WorkflowJob.pyi
@@ -0,0 +1,39 @@
from datetime import datetime
from typing import Any, Dict, List

from github.GithubObject import CompletableGithubObject
from github.WorkflowStep import WorkflowStep

class WorkflowJob(CompletableGithubObject):
def __repr__(self) -> str: ...
def _initAttributes(self) -> None: ...
def _useAttributes(self, attributes: Dict[str, Any]) -> None: ...
@property
def check_run_url(self) -> str: ...
@property
def completed_at(self) -> datetime: ...
@property
def conclusion(self) -> str: ...
@property
def head_sha(self) -> str: ...
@property
def html_url(self) -> str: ...
@property
def id(self) -> int: ...
@property
def name(self) -> str: ...
@property
def node_id(self) -> str: ...
@property
def run_id(self) -> int: ...
@property
def run_url(self) -> str: ...
@property
def started_at(self) -> datetime: ...
@property
def status(self) -> str: ...
@property
def steps(self) -> List[WorkflowStep]: ...
@property
def url(self) -> str: ...
def logs_url(self) -> str: ...
23 changes: 23 additions & 0 deletions github/WorkflowRun.py
Expand Up @@ -26,6 +26,7 @@
import github.Artifact
import github.GithubObject
import github.PullRequest
import github.WorkflowJob


class WorkflowRun(github.GithubObject.CompletableGithubObject):
Expand Down Expand Up @@ -302,6 +303,28 @@ def delete(self):
status, _, _ = self._requester.requestJson("DELETE", self.url)
return status == 204

def jobs(self, _filter=github.GithubObject.NotSet):
"""
:calls "`GET /repos/{owner}/{repo}/actions/runs/{run_id}/jobs <https://docs.github.com/en/rest/reference/actions#list-jobs-for-a-workflow-run>`_
:param _filter: string `latest`, or `all`
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.WorkflowJob.WorkflowJob`
"""
assert _filter is github.GithubObject.NotSet or isinstance(
_filter, str
), _filter

url_parameters = dict()
if _filter is not github.GithubObject.NotSet:
url_parameters["filter"] = _filter

return github.PaginatedList.PaginatedList(
github.WorkflowJob.WorkflowJob,
self._requester,
self.jobs_url,
url_parameters,
list_item="jobs",
)

def _initAttributes(self):
self._id = github.GithubObject.NotSet
self._name = github.GithubObject.NotSet
Expand Down
9 changes: 7 additions & 2 deletions github/WorkflowRun.pyi
@@ -1,10 +1,12 @@
from datetime import datetime
from typing import Any, Dict, NamedTuple, List
from typing import Any, Dict, NamedTuple, List, Union

from github.GitCommit import GitCommit
from github.GithubObject import CompletableGithubObject
from github.GithubObject import CompletableGithubObject, _NotSetType
from github.PaginatedList import PaginatedList
from github.PullRequest import PullRequest
from github.Repository import Repository
from github.WorkflowJob import WorkflowJob

class TimingData(NamedTuple):
billable: Dict[str, Dict[str, int]]
Expand Down Expand Up @@ -71,5 +73,8 @@ class WorkflowRun(CompletableGithubObject):
@property
def head_repository(self) -> Repository: ...
def cancel(self) -> bool: ...
def jobs(
self, _filter: Union[str, _NotSetType] = ...
) -> PaginatedList[WorkflowJob]: ...
def rerun(self) -> bool: ...
def timing(self) -> TimingData: ...

0 comments on commit 804c310

Please sign in to comment.