forked from sigmavirus24/github3.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrelease.py
More file actions
219 lines (188 loc) · 7.94 KB
/
release.py
File metadata and controls
219 lines (188 loc) · 7.94 KB
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from ..decorators import requires_auth
from ..models import GitHubCore, GitHubError
from ..utils import stream_response_to_file
from uritemplate import URITemplate
class Release(GitHubCore):
"""The :class:`Release <Release>` object.
It holds the information GitHub returns about a release from a
:class:`Repository <github3.repos.repo.Repository>`.
"""
CUSTOM_HEADERS = {'Accept': 'application/vnd.github.manifold-preview'}
def __init__(self, release, session=None):
super(Release, self).__init__(release, session)
self._api = release.get('url')
#: List of :class:`Asset <Asset>` objects for this release
self.assets = [Asset(i, self) for i in release.get('assets', [])]
#: URL for uploaded assets
self.assets_url = release.get('assets_url')
#: Body of the release (the description)
self.body = release.get('body')
#: Date the release was created
self.created_at = self._strptime(release.get('created_at'))
#: Boolean whether value is True or False
self.draft = release.get('draft')
#: HTML URL of the release
self.html_url = release.get('html_url')
#: GitHub id
self.id = release.get('id')
#: Name given to the release
self.name = release.get('name')
#; Boolean whether release is a prerelease
self.prerelease = release.get('prerelease')
#: Date the release was published
self.published_at = self._strptime(release.get('published_at'))
#: Name of the tag
self.tag_name = release.get('tag_name')
#: "Commit" that this release targets
self.target_commitish = release.get('target_commitish')
upload_url = release.get('upload_url')
#: URITemplate to upload an asset with
self.upload_urlt = URITemplate(upload_url) if upload_url else None
def _repr(self):
return '<Release [{0}]>'.format(self.name)
@requires_auth
def delete(self):
"""Users with push access to the repository can delete a release.
:returns: True if successful; False if not successful
"""
url = self._api
return self._boolean(
self._delete(url, headers=Release.CUSTOM_HEADERS),
204,
404
)
@requires_auth
def edit(self, tag_name=None, target_commitish=None, name=None, body=None,
draft=None, prerelease=None):
"""Users with push access to the repository can edit a release.
If the edit is successful, this object will update itself.
:param str tag_name: (optional), Name of the tag to use
:param str target_commitish: (optional), The "commitish" value that
determines where the Git tag is created from. Defaults to the
repository's default branch.
:param str name: (optional), Name of the release
:param str body: (optional), Description of the release
:param boolean draft: (optional), True => Release is a draft
:param boolean prerelease: (optional), True => Release is a prerelease
:returns: True if successful; False if not successful
"""
url = self._api
data = {
'tag_name': tag_name,
'target_commitish': target_commitish,
'name': name,
'body': body,
'draft': draft,
'prerelease': prerelease,
}
self._remove_none(data)
r = self._session.patch(
url, data=json.dumps(data), headers=Release.CUSTOM_HEADERS
)
successful = self._boolean(r, 200, 404)
if successful:
# If the edit was successful, let's update the object.
self.__init__(r.json(), self)
return successful
def iter_assets(self, number=-1, etag=None):
"""Iterate over the assets available for this release.
:param int number: (optional), Number of assets to return
:param str etag: (optional), last ETag header sent
:returns: generator of :class:`Asset <Asset>` objects
"""
url = self._build_url('assets', base_url=self._api)
return self._iter(number, url, Asset, etag=etag)
@requires_auth
def upload_asset(self, content_type, name, asset):
"""Upload an asset to this release.
All parameters are required.
:param str content_type: The content type of the asset. Wikipedia has
a list of common media types
:param str name: The name of the file
:param asset: The file or bytes object to upload.
:returns: :class:`Asset <Asset>`
"""
headers = Release.CUSTOM_HEADERS.copy()
headers.update({'Content-Type': content_type})
url = self.upload_urlt.expand({'name': name})
r = self._post(url, data=asset, json=False, headers=headers,
verify=False)
if r.status_code in (201, 202):
return Asset(r.json(), self)
raise GitHubError(r)
class Asset(GitHubCore):
def __init__(self, asset, session=None):
super(Asset, self).__init__(asset, session)
self._api = asset.get('url')
#: Content-Type provided when the asset was created
self.content_type = asset.get('content_type')
#: Date the asset was created
self.created_at = self._strptime(asset.get('created_at'))
#: Number of times the asset was downloaded
self.download_count = asset.get('download_count')
#: URL to download the asset.
#: Request headers must include ``Accept: application/octet-stream``.
self.download_url = self._api
#: GitHub id of the asset
self.id = asset.get('id')
#: Short description of the asset
self.label = asset.get('label')
#: Name of the asset
self.name = asset.get('name')
#: Size of the asset
self.size = asset.get('size')
#: State of the asset, e.g., "uploaded"
self.state = asset.get('state')
#: Date the asset was updated
self.updated_at = self._strptime(asset.get('updated_at'))
def _repr(self):
return '<Asset [{0}]>'.format(self.name)
def download(self, path=''):
"""Download the data for this asset.
:param path: (optional), path where the file should be saved
to, default is the filename provided in the headers and will be
written in the current directory.
it can take a file-like object as well
:type path: str, file
:returns: bool -- True if successful, False otherwise
"""
headers = {
'Accept': 'application/octet-stream'
}
resp = self._get(self._api, allow_redirects=False, stream=True,
headers=headers)
if resp.status_code == 302:
# Amazon S3 will reject the redirected request unless we omit
# certain request headers
headers.update({
'Content-Type': None,
})
with self._session.no_auth():
resp = self._get(resp.headers['location'], stream=True,
headers=headers)
if self._boolean(resp, 200, 404):
stream_response_to_file(resp, path)
return True
return False
def edit(self, name, label=None):
"""Edit this asset.
:param str name: (required), The file name of the asset
:param str label: (optional), An alternate description of the asset
:returns: boolean
"""
if not name:
return False
edit_data = {'name': name, 'label': label}
self._remove_none(edit_data)
r = self._patch(
self._api,
data=json.dumps(edit_data),
headers=Release.CUSTOM_HEADERS
)
successful = self._boolean(r, 200, 404)
if successful:
self.__init__(r.json(), self)
return successful