python-gitlab/gitlab/utils.py at main · srini387/python-gitlab · GitHub

Repository navigation

History

292 lines (228 loc) · 8.97 KB

Raw

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

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 = 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 not transform_data:

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

python-gitlab/gitlab/utils.py at main · srini387/python-gitlab · GitHub

Repository navigation

Expand file tree

Breadcrumbs

utils.py

Latest commit

History

Breadcrumbs

utils.py

File metadata and controls