Move requests handling to corelibs_requests module
This commit is contained in:
Clemens Schwaighofer
@@ -21,7 +21,7 @@ def mask(
|
||||
mask_str: str = "***",
|
||||
mask_str_edges: str = '_',
|
||||
skip: bool = False
|
||||
) -> dict[str, Any]:
|
||||
) -> dict[str, Any] | list[Any]:
|
||||
"""
|
||||
mask data for output
|
||||
Checks if mask_keys list exist in any key in the data set either from the start or at the end
|
||||
|
||||
@@ -2,9 +2,11 @@
|
||||
Various HTTP auth helpers
|
||||
"""
|
||||
|
||||
from base64 import b64encode
|
||||
from warnings import deprecated
|
||||
from corelibs_requests.auth_helpers import basic_auth as corelibs_basic_auth
|
||||
|
||||
|
||||
@deprecated("use corelibs_requests.auth_helpers.basic_auth instead")
|
||||
def basic_auth(username: str, password: str) -> str:
|
||||
"""
|
||||
setup basic auth, for debug
|
||||
@@ -16,5 +18,6 @@ def basic_auth(username: str, password: str) -> str:
|
||||
Returns:
|
||||
str -- _description_
|
||||
"""
|
||||
token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii")
|
||||
return f'Basic {token}'
|
||||
return corelibs_basic_auth(username, password)
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -3,264 +3,35 @@ requests lib interface
|
||||
V2 call type
|
||||
"""
|
||||
|
||||
from typing import Any, TypedDict, cast
|
||||
import requests
|
||||
from requests import exceptions
|
||||
from warnings import warn
|
||||
from corelibs_requests.caller import (
|
||||
Caller as CoreLibsCaller,
|
||||
ProxyConfig as CoreLibsProxyConfig,
|
||||
ErrorResponse as CoreLibsErrorResponse
|
||||
)
|
||||
|
||||
|
||||
class ErrorResponse:
|
||||
class ErrorResponse(CoreLibsErrorResponse):
|
||||
"""
|
||||
Error response structure. This is returned if a request could not be completed
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
code: int,
|
||||
message: str,
|
||||
action: str,
|
||||
url: str,
|
||||
exception: exceptions.InvalidSchema | exceptions.ReadTimeout | exceptions.ConnectionError | None = None
|
||||
) -> None:
|
||||
self.code = code
|
||||
self.message = message
|
||||
self.action = action
|
||||
self.url = url
|
||||
self.exception_name = type(exception).__name__ if exception is not None else None
|
||||
self.exception_trace = exception if exception is not None else None
|
||||
|
||||
|
||||
class ProxyConfig(TypedDict):
|
||||
class ProxyConfig(CoreLibsProxyConfig):
|
||||
"""
|
||||
Socks proxy settings
|
||||
"""
|
||||
type: str
|
||||
host: str
|
||||
port: str
|
||||
|
||||
|
||||
class Caller:
|
||||
class Caller(CoreLibsCaller):
|
||||
"""
|
||||
requests lib interface
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
header: dict[str, str],
|
||||
timeout: int = 20,
|
||||
proxy: ProxyConfig | None = None,
|
||||
verify: bool = True,
|
||||
ca_file: str | None = None
|
||||
):
|
||||
self.headers = header
|
||||
self.timeout: int = timeout
|
||||
self.ca_file = ca_file
|
||||
self.verify = verify
|
||||
self.proxy = cast(dict[str, str], proxy) if proxy is not None else None
|
||||
|
||||
def __timeout(self, timeout: int | None) -> int:
|
||||
if timeout is not None and timeout >= 0:
|
||||
return timeout
|
||||
return self.timeout
|
||||
|
||||
def __call(
|
||||
self,
|
||||
action: str,
|
||||
url: str,
|
||||
data: dict[str, Any] | None = None,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""
|
||||
call wrapper, on error returns None
|
||||
|
||||
Args:
|
||||
action (str): _description_
|
||||
url (str): _description_
|
||||
data (dict | None): _description_. Defaults to None.
|
||||
params (dict | None): _description_. Defaults to None.
|
||||
|
||||
Returns:
|
||||
requests.Response | None: _description_
|
||||
"""
|
||||
|
||||
if data is None:
|
||||
data = {}
|
||||
try:
|
||||
if action == "get":
|
||||
return requests.get(
|
||||
url,
|
||||
params=params,
|
||||
headers=self.headers,
|
||||
timeout=self.__timeout(timeout),
|
||||
verify=self.verify,
|
||||
proxies=self.proxy,
|
||||
cert=self.ca_file
|
||||
)
|
||||
if action == "post":
|
||||
return requests.post(
|
||||
url,
|
||||
params=params,
|
||||
json=data,
|
||||
headers=self.headers,
|
||||
timeout=self.__timeout(timeout),
|
||||
verify=self.verify,
|
||||
proxies=self.proxy,
|
||||
cert=self.ca_file
|
||||
)
|
||||
if action == "put":
|
||||
return requests.put(
|
||||
url,
|
||||
params=params,
|
||||
json=data,
|
||||
headers=self.headers,
|
||||
timeout=self.__timeout(timeout),
|
||||
verify=self.verify,
|
||||
proxies=self.proxy,
|
||||
cert=self.ca_file
|
||||
)
|
||||
if action == "patch":
|
||||
return requests.patch(
|
||||
url,
|
||||
params=params,
|
||||
json=data,
|
||||
headers=self.headers,
|
||||
timeout=self.__timeout(timeout),
|
||||
verify=self.verify,
|
||||
proxies=self.proxy,
|
||||
cert=self.ca_file
|
||||
)
|
||||
if action == "delete":
|
||||
return requests.delete(
|
||||
url,
|
||||
params=params,
|
||||
headers=self.headers,
|
||||
timeout=self.__timeout(timeout),
|
||||
verify=self.verify,
|
||||
proxies=self.proxy,
|
||||
cert=self.ca_file
|
||||
)
|
||||
return ErrorResponse(
|
||||
100,
|
||||
f"Unsupported action '{action}'",
|
||||
action,
|
||||
url
|
||||
)
|
||||
except exceptions.InvalidSchema as e:
|
||||
return ErrorResponse(
|
||||
200,
|
||||
f"Invalid URL during '{action}' for {url}",
|
||||
action,
|
||||
url,
|
||||
e
|
||||
)
|
||||
except exceptions.ReadTimeout as e:
|
||||
return ErrorResponse(
|
||||
300,
|
||||
f"Timeout ({self.timeout}s) during '{action}' for {url}",
|
||||
action,
|
||||
url,
|
||||
e
|
||||
)
|
||||
except exceptions.ConnectionError as e:
|
||||
return ErrorResponse(
|
||||
400,
|
||||
f"Connection error during '{action}' for {url}",
|
||||
action,
|
||||
url,
|
||||
e
|
||||
)
|
||||
|
||||
def get(
|
||||
self,
|
||||
url: str,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""
|
||||
get data
|
||||
|
||||
Args:
|
||||
url (str): _description_
|
||||
params (dict | None): _description_
|
||||
|
||||
Returns:
|
||||
requests.Response: _description_
|
||||
"""
|
||||
return self.__call('get', url, params=params, timeout=timeout)
|
||||
|
||||
def post(
|
||||
self,
|
||||
url: str,
|
||||
data: dict[str, Any] | None = None,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""
|
||||
post data
|
||||
|
||||
Args:
|
||||
url (str): _description_
|
||||
data (dict | None): _description_
|
||||
params (dict | None): _description_
|
||||
|
||||
Returns:
|
||||
requests.Response | None: _description_
|
||||
"""
|
||||
return self.__call('post', url, data, params, timeout=timeout)
|
||||
|
||||
def put(
|
||||
self,
|
||||
url: str,
|
||||
data: dict[str, Any] | None = None,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
url (str): _description_
|
||||
data (dict | None): _description_
|
||||
params (dict | None): _description_
|
||||
|
||||
Returns:
|
||||
requests.Response | None: _description_
|
||||
"""
|
||||
return self.__call('put', url, data, params, timeout=timeout)
|
||||
|
||||
def patch(
|
||||
self,
|
||||
url: str,
|
||||
data: dict[str, Any] | None = None,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""_summary_
|
||||
|
||||
Args:
|
||||
url (str): _description_
|
||||
data (dict | None): _description_
|
||||
params (dict | None): _description_
|
||||
|
||||
Returns:
|
||||
requests.Response | None: _description_
|
||||
"""
|
||||
return self.__call('patch', url, data, params, timeout=timeout)
|
||||
|
||||
def delete(
|
||||
self,
|
||||
url: str,
|
||||
params: dict[str, Any] | None = None,
|
||||
timeout: int | None = None
|
||||
) -> requests.Response | ErrorResponse:
|
||||
"""
|
||||
delete
|
||||
|
||||
Args:
|
||||
url (str): _description_
|
||||
params (dict | None): _description_
|
||||
|
||||
Returns:
|
||||
requests.Response | None: _description_
|
||||
"""
|
||||
return self.__call('delete', url, params=params, timeout=timeout)
|
||||
warn(
|
||||
"corelibs.requests_handling.caller is deprecated, use corelibs_requests.caller instead",
|
||||
DeprecationWarning, stacklevel=2
|
||||
)
|
||||
|
||||
# __END__
|
||||
|
||||
Reference in New Issue
Block a user