Repository navigation
Expand file tree
/
Copy pathutils.py
More file actions
300 lines (236 loc) · 9.54 KB
/
Copy pathutils.py
File metadata and controls
300 lines (236 loc) · 9.54 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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
from __future__ import annotations
import dataclasses
import email.message
import logging
import pathlib
import time
import traceback
import urllib.parse
import warnings
from collections.abc import Iterator, MutableMapping
from typing import Any, Callable, Literal
import requests
from gitlab import const, types
class _StdoutStream:
def __call__(self, chunk: Any) -> None:
print(chunk)
def get_base_url(url: str | None = None) -> str:
"""Return the base URL with the trailing slash stripped.
If the URL is a Falsy value, return the default URL.
Returns:
The base URL
"""
if not url:
return const.DEFAULT_URL
return url.rstrip("/")
def get_content_type(content_type: str | None) -> str:
message = email.message.Message()
if content_type is not None:
message["content-type"] = content_type
return message.get_content_type()
class MaskingFormatter(logging.Formatter):
"""A logging formatter that can mask credentials"""
def __init__(
self,
fmt: str | None = logging.BASIC_FORMAT,
datefmt: str | None = None,
style: Literal["%", "{", "$"] = "%",
validate: bool = True,
masked: str | None = None,
) -> None:
super().__init__(fmt, datefmt, style, validate)
self.masked = masked
def _filter(self, entry: str) -> str:
if not self.masked:
return entry
return entry.replace(self.masked, "[MASKED]")
def format(self, record: logging.LogRecord) -> str:
original = logging.Formatter.format(self, record)
return self._filter(original)
def response_content(
response: requests.Response,
streamed: bool,
action: Callable[[bytes], Any] | None,
chunk_size: int,
*,
iterator: bool,
) -> bytes | Iterator[Any] | None:
if iterator:
return response.iter_content(chunk_size=chunk_size)
if streamed is False:
return response.content
if action is None:
action = _StdoutStream()
for chunk in response.iter_content(chunk_size=chunk_size):
if chunk:
action(chunk)
return None
class Retry:
def __init__(
self,
max_retries: int,
obey_rate_limit: bool | None = True,
retry_transient_errors: bool | None = False,
) -> None:
self.cur_retries = 0
self.max_retries = max_retries
self.obey_rate_limit = obey_rate_limit
self.retry_transient_errors = retry_transient_errors
def _retryable_status_code(self, status_code: int | None, reason: str = "") -> bool:
if status_code == 429 and self.obey_rate_limit:
return True
if not self.retry_transient_errors:
return False
if status_code in const.RETRYABLE_TRANSIENT_ERROR_CODES:
return True
if status_code == 409 and "Resource lock" in reason:
return True
return False
def handle_retry_on_status(
self,
status_code: int | None,
headers: MutableMapping[str, str] | None = None,
reason: str = "",
) -> bool:
if not self._retryable_status_code(status_code, reason):
return False
if headers is None:
headers = {}
# Response headers documentation:
# https://docs.gitlab.com/ee/user/admin_area/settings/user_and_ip_rate_limits.html#response-headers
if self.max_retries == -1 or self.cur_retries < self.max_retries:
wait_time = 2**self.cur_retries * 0.1
if "Retry-After" in headers:
wait_time = int(headers["Retry-After"])
elif "RateLimit-Reset" in headers:
wait_time = max(0, int(headers["RateLimit-Reset"]) - time.time())
self.cur_retries += 1
time.sleep(wait_time)
return True
return False
def handle_retry(self) -> bool:
if self.retry_transient_errors and (
self.max_retries == -1 or self.cur_retries < self.max_retries
):
wait_time = 2**self.cur_retries * 0.1
self.cur_retries += 1
time.sleep(wait_time)
return True
return False
def _transform_types(
data: dict[str, Any],
custom_types: dict[str, Any],
*,
transform_data: bool,
transform_files: bool | None = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Copy the data dict with attributes that have custom types and transform them
before being sent to the server.
``transform_files``: If ``True`` (default), also populates the ``files`` dict for
FileAttribute types with tuples to prepare fields for requests' MultipartEncoder:
https://toolbelt.readthedocs.io/en/latest/user.html#multipart-form-data-encoder
``transform_data``: If ``True`` transforms the ``data`` dict with fields
suitable for encoding as query parameters for GitLab's API:
https://docs.gitlab.com/ee/api/#encoding-api-parameters-of-array-and-hash-types
Returns:
A tuple of the transformed data dict and files dict"""
# Duplicate data to avoid messing with what the user sent us
data = data.copy()
if not transform_files and not transform_data:
return data, {}
files = {}
for attr_name, attr_class in custom_types.items():
if attr_name not in data:
continue
gitlab_attribute = attr_class(data[attr_name])
# if the type is FileAttribute we need to pass the data as file
if isinstance(gitlab_attribute, types.FileAttribute) and transform_files:
# The GitLab API accepts mixed types
# (e.g. a file for avatar image or empty string for removing the avatar)
# So if string is empty, keep it in data dict
if isinstance(data[attr_name], str) and data[attr_name] == "":
continue
key = gitlab_attribute.get_file_name(attr_name)
files[attr_name] = (key, data.pop(attr_name))
continue
# If transform_data is False, it means we are preparing data for a JSON body
# (POST/PUT/PATCH). In this case, we normally skip transformation because
# most types (like ArrayAttribute) only need transformation for query
# parameters (GET).
#
# However, some types (like CommaSeparatedStringAttribute) need to be
# transformed even in JSON bodies (e.g. converting a list to a CSV string).
# The 'transform_in_body' flag on the attribute class controls this behavior.
if not transform_data and not gitlab_attribute.transform_in_body:
continue
if isinstance(gitlab_attribute, types.GitlabAttribute):
key, value = gitlab_attribute.get_for_api(key=attr_name)
if key != attr_name:
del data[attr_name]
data[key] = value
return data, files
def copy_dict(*, src: dict[str, Any], dest: dict[str, Any]) -> None:
for k, v in src.items():
if isinstance(v, dict):
# NOTE(jlvillal): This provides some support for the `hash` type
# https://docs.gitlab.com/ee/api/#hash
# Transform dict values to new attributes. For example:
# custom_attributes: {'foo', 'bar'} =>
# "custom_attributes['foo']": "bar"
for dict_k, dict_v in v.items():
dest[f"{k}[{dict_k}]"] = dict_v
else:
dest[k] = v
class EncodedId(str):
"""A custom `str` class that will return the URL-encoded value of the string.
* Using it recursively will only url-encode the value once.
* Can accept either `str` or `int` as input value.
* Can be used in an f-string and output the URL-encoded string.
Reference to documentation on why this is necessary.
See::
https://docs.gitlab.com/ee/api/index.html#namespaced-path-encoding
https://docs.gitlab.com/ee/api/index.html#path-parameters
"""
def __new__(cls, value: str | int | EncodedId) -> EncodedId:
if isinstance(value, EncodedId):
return value
if not isinstance(value, (int, str)):
raise TypeError(f"Unsupported type received: {type(value)}")
if isinstance(value, str):
value = urllib.parse.quote(value, safe="")
return super().__new__(cls, value)
def remove_none_from_dict(data: dict[str, Any]) -> dict[str, Any]:
return {k: v for k, v in data.items() if v is not None}
def warn(
message: str,
*,
category: type[Warning] | None = None,
source: Any | None = None,
show_caller: bool = True,
) -> None:
"""This `warnings.warn` wrapper function attempts to show the location causing the
warning in the user code that called the library.
It does this by walking up the stack trace to find the first frame located outside
the `gitlab/` directory. This is helpful to users as it shows them their code that
is causing the warning.
"""
# Get `stacklevel` for user code so we indicate where issue is in
# their code.
pg_dir = pathlib.Path(__file__).parent.resolve()
stack = traceback.extract_stack()
stacklevel = 1
warning_from = ""
for stacklevel, frame in enumerate(reversed(stack), start=1):
warning_from = f" (python-gitlab: {frame.filename}:{frame.lineno})"
frame_dir = str(pathlib.Path(frame.filename).parent.resolve())
if not frame_dir.startswith(str(pg_dir)):
break
if show_caller:
message += warning_from
warnings.warn(
message=message, category=category, stacklevel=stacklevel, source=source
)
@dataclasses.dataclass
class WarnMessageData:
message: str
show_caller: bool