forked from pypa/twine
-
Notifications
You must be signed in to change notification settings - Fork 0
/
upload.py
208 lines (165 loc) · 7.51 KB
/
upload.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
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
"""Module containing the upload function in twine.
This module uploads the package to the repository.
"""
# Copyright 2013 Donald Stufft
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import logging
import os.path
from typing import Dict, List, cast
import requests
from twine import commands
from twine import exceptions
from twine import package as package_file
from twine import settings
from twine import utils
logger = logging.getLogger(__name__)
def skip_upload(
response: requests.Response, skip_existing: bool, package: package_file.PackageFile
) -> bool:
"""Skip uploading a package.
Return Boolean type according to the status code that responded by the repository or
the argument passed by the user when trying to upload the package(s).
If ``skip_existing`` is set to ``True``, then return ``False``.
If status code 400, 403, 409 is responded by the repository, return ``True``.
:param requests.Response response:
Get the response from the repository.
:param bool skip_existing:
Specify whether twine should continue uploading files if one
of them already exists. This primarily supports PyPI. Other
package indexes may not be supported.
:param package_file.PackageFile package:
Get the package files.
:return bool:
Determine whether we should skip uploading the package.
"""
if not skip_existing:
return False
status = response.status_code
reason = getattr(response, "reason", "").lower()
text = getattr(response, "text", "").lower()
# NOTE(sigmavirus24): PyPI presently returns a 400 status code with the
# error message in the reason attribute. Other implementations return a
# 403 or 409 status code.
return (
# pypiserver (https://pypi.org/project/pypiserver)
status == 409
# PyPI / TestPyPI
or (status == 400 and "already exist" in reason)
# Nexus Repository OSS (https://www.sonatype.com/nexus-repository-oss)
or (status == 400 and any("updating asset" in x for x in [reason, text]))
# Artifactory (https://jfrog.com/artifactory/)
or (status == 403 and "overwrite artifact" in text)
# Gitlab Enterprise Edition (https://about.gitlab.com)
or (status == 400 and "already been taken" in text)
)
def _make_package(
filename: str, signatures: Dict[str, str], upload_settings: settings.Settings
) -> package_file.PackageFile:
package = package_file.PackageFile.from_filename(filename, upload_settings.comment)
signed_name = package.signed_basefilename
if signed_name in signatures:
package.add_gpg_signature(signatures[signed_name], signed_name)
elif upload_settings.sign:
package.sign(upload_settings.sign_with, upload_settings.identity)
file_size = utils.get_file_size(package.filename)
logger.info(f" {package.filename} ({file_size})")
if package.gpg_signature:
logger.info(f" Signed with {package.signed_filename}")
return package
def upload(upload_settings: settings.Settings, dists: List[str]) -> None:
"""Upload distributions to repository.
This function will check if the user has passed in pre-signed distributions.
Then, it will check the repository url to verify that we are not using legacy
PyPI. If no error is occurred, it will make package, create repository and
upload distributions.
If ``skip_existing`` is set to ``True`` and the package is uploaded already,
it prints the skipping message to the user and continues uploading distributions.
If we get a redirect, exception :class:`RedirectDetected` is raised.
If ``skip_upload`` is ``True``, it prints the skipping message to the user
and continues uploading distributions.
Then, it will check status code responded by the repository, and generate
a helpful message. After that, it will add the distribution files uploaded to
the ``uploaded_packages``.
Finally, it will show release urls to the user and close the session.
:param settings.Settings upload_settings:
The settings for the upload function.
:param List[str] dists:
Get dists that are going to be uploaded.
"""
dists = commands._find_dists(dists)
# Determine if the user has passed in pre-signed distributions
signatures = {os.path.basename(d): d for d in dists if d.endswith(".asc")}
uploads = [i for i in dists if not i.endswith(".asc")]
upload_settings.check_repository_url()
repository_url = cast(str, upload_settings.repository_config["repository"])
print(f"Uploading distributions to {repository_url}")
packages_to_upload = [
_make_package(filename, signatures, upload_settings) for filename in uploads
]
repository = upload_settings.create_repository()
uploaded_packages = []
for package in packages_to_upload:
skip_message = " Skipping {} because it appears to already exist".format(
package.basefilename
)
# Note: The skip_existing check *needs* to be first, because otherwise
# we're going to generate extra HTTP requests against a hardcoded
# URL for no reason.
if upload_settings.skip_existing and repository.package_is_uploaded(package):
print(skip_message)
continue
resp = repository.upload(package)
# Bug 92. If we get a redirect we should abort because something seems
# funky. The behaviour is not well defined and redirects being issued
# by PyPI should never happen in reality. This should catch malicious
# redirects as well.
if resp.is_redirect:
raise exceptions.RedirectDetected.from_args(
repository_url,
resp.headers["location"],
)
if skip_upload(resp, upload_settings.skip_existing, package):
print(skip_message)
continue
utils.check_status_code(resp, upload_settings.verbose)
uploaded_packages.append(package)
release_urls = repository.release_urls(uploaded_packages)
if release_urls:
print("\nView at:")
for url in release_urls:
print(url)
# Bug 28. Try to silence a ResourceWarning by clearing the connection
# pool.
repository.close()
def main(args: List[str]) -> None:
"""Entry-point of upload command.
:param List[str] args:
Arguments for the upload command.
"""
parser = argparse.ArgumentParser(prog="twine upload")
settings.Settings.register_argparse_arguments(parser)
parser.add_argument(
"dists",
nargs="+",
metavar="dist",
help="The distribution files to upload to the repository "
"(package index). Usually dist/* . May additionally contain "
"a .asc file to include an existing signature with the "
"file upload.",
)
parsed_args = parser.parse_args(args)
upload_settings = settings.Settings.from_argparse(parsed_args)
# Call the upload function with the arguments from the command line
return upload(upload_settings, parsed_args.dists)