Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for workflow jobs and steps #1951

Merged
merged 4 commits into from Jun 20, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
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):
EnricoMi marked this conversation as resolved.
Show resolved Hide resolved
"""
:type: string
"""
print(f"{self.url}/logs")
EnricoMi marked this conversation as resolved.
Show resolved Hide resolved
headers, _ = self._requester.requestBlobAndCheck("GET", f"{self.url}/logs")
return headers["location"]
EnricoMi marked this conversation as resolved.
Show resolved Hide resolved

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: ...