re-add normalize_scope and Retry class to util.py

This commit is contained in:
dieser-niko 2024-10-13 21:59:36 +02:00
parent 701dc383ea
commit 364d082396

View File

@ -38,3 +38,53 @@ def get_host_port(netloc):
port = None
return host, port
def normalize_scope(scope):
"""Normalize the scope to verify that it is a list or tuple. A string
input will split the string by commas to create a list of scopes.
A list or tuple input is used directly.
Parameters:
- scope - a string representing scopes separated by commas,
or a list/tuple of scopes.
"""
if scope:
if isinstance(scope, str):
scopes = scope.split(',')
elif isinstance(scope, list) or isinstance(scope, tuple):
scopes = scope
else:
raise Exception(
"Unsupported scope value, please either provide a list of scopes, "
"or a string of scopes separated by commas."
)
return " ".join(sorted(scopes))
else:
return None
class Retry(urllib3.Retry):
"""
Custom class for printing a warning when a rate/request limit is reached.
"""
def increment(
self,
method: str | None = None,
url: str | None = None,
response: urllib3.BaseHTTPResponse | None = None,
error: Exception | None = None,
_pool: urllib3.connectionpool.ConnectionPool | None = None,
_stacktrace: TracebackType | None = None,
) -> urllib3.Retry:
if response:
retry_header = response.headers.get("Retry-After")
if self.is_retry(method, response.status, bool(retry_header)):
logging.warning("Your application has reached a rate/request limit. "
f"Retry will occur after: {retry_header}")
return super().increment(method,
url,
response=response,
error=error,
_pool=_pool,
_stacktrace=_stacktrace)