-
Notifications
You must be signed in to change notification settings - Fork 1.6k
/
prs.py
94 lines (82 loc) · 3.5 KB
/
prs.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import json
from typing import Any
from prefect import Task
from prefect.utilities.tasks import defaults_from_attrs
class CreateGitHubPR(Task):
"""
Task for opening/creating new GitHub Pull Requests using the v3 version of the GitHub REST API.
Args:
- repo (str, optional): the name of the repository to open the issue in; must be
provided in the form `organization/repo_name`; can also be provided to the `run`
method
- title (str, optional): the title of the issue to create; can also be provided to the
`run` method
- body (str, optional): the contents of the issue; can also be provided to the `run` method
- head (str, optional): the name of the branch where your changes are implemented; can also
be provided to the `run` method
- base (str, optional): the name of the branch you want the changes pulled into; can also
be provided to the `run` method
- **kwargs (Any, optional): additional keyword arguments to pass to the standard Task
init method
"""
def __init__(
self,
repo: str = None,
title: str = None,
body: str = None,
head: str = None,
base: str = None,
**kwargs: Any
):
self.repo = repo
self.title = title
self.body = body
self.head = head
self.base = base
super().__init__(**kwargs)
@defaults_from_attrs("repo", "title", "body", "head", "base")
def run(
self,
repo: str = None,
title: str = None,
body: str = None,
head: str = None,
base: str = None,
token: str = None,
) -> None:
"""
Run method for this Task. Invoked by calling this Task after initialization within a
Flow context, or by using `Task.bind`.
Args:
- repo (str, optional): the name of the repository to open the issue in; must be
provided in the form `organization/repo_name`; defaults to the one provided at
initialization
- title (str, optional): the title of the issue to create; defaults to the one
provided at initialization
- body (str, optional): the contents of the issue; defaults to the one provided at
initialization
- head (str, optional): the name of the branch where your changes are implemented;
defaults to the one provided at initialization
- base (str, optional): the name of the branch you want the changes pulled into;
defaults to the one provided at initialization
- token (str): a GitHub API token
Raises:
- ValueError: if a `repo` was never provided
- HTTPError: if the POST request returns a non-200 status code
Returns:
- None
"""
if repo is None:
raise ValueError("A GitHub repository must be provided.")
# 'import requests' is expensive time-wise, we should do this just-in-time to keep
# the 'import prefect' time low
import requests
url = "https://api.github.com/repos/{}/pulls".format(repo)
headers = {
"AUTHORIZATION": "token {}".format(token),
"Accept": "application/vnd.github.v3+json",
}
pr = {"title": title, "body": body, "head": head, "base": base}
# send the request
resp = requests.post(url, data=json.dumps(pr), headers=headers)
resp.raise_for_status()