Skip to content

Adapter

Request adapter performing actual API calls towards FIO endpoints

logger module-attribute

logger = logging.getLogger(__name__)

FIOAdapter

FIOAdapter(config, header)

FIO Adapater

Initializes the FIO adapter

Parameters:

Name Type Description Default
config Config

Wrapper configuration.

required
header Dict[str, str]

FIO Header.

required
Source code in fio_wrapper/fio_adapter.py
def __init__(self, config: Config, header: Dict[str, str]):
    """Initializes the FIO adapter

    Args:
        config (Config): Wrapper configuration.
        header (Dict[str, str]): FIO Header.
    """

    self.config = config
    self.header = header
    self.ssl_verify = self.config.ssl_verify
    self.timeout = self.config.timeout

    if not self.ssl_verify:
        requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

    # register a session based on requests.session or, if installed
    # requests-cache session handler

    if self.config.cache and importlib.util.find_spec("requests_cache") is not None:
        logger.debug("Using requests-cache Session")
        from requests_cache import CachedSession

        self._session = CachedSession(
            cache_name="fio_wrapper",
            backend="memory",
            expire_after=self.config.cache_default_expire,
            urls_expire_after=self.config.cache_url_expirations(),
            cache_control=False,
        )
    else:
        logger.debug("Using request Session")
        self._session = requests.session()

config instance-attribute

config = config

header instance-attribute

header = header

ssl_verify instance-attribute

ssl_verify = self.config.ssl_verify

timeout instance-attribute

timeout = self.config.timeout

get

get(endpoint, params=None, err_codes=None, timeout=None)

Performs a GET request towards endpoint

Parameters:

Name Type Description Default
endpoint str

URL

required
params Dict

GET parameters. Defaults to None.

None
err_codes List[int]

List of error codes to handle in calling function. Defaults to None.

None
timeout float

Request timeout in seconds. Defaults to None.

None

Returns:

Type Description
Tuple[int, any]

Tuple[int, any]: Request status code and request data

Source code in fio_wrapper/fio_adapter.py
def get(
    self,
    endpoint: str,
    params: Dict = None,
    err_codes: Optional[List[int]] = None,
    timeout: Optional[float] = None,
) -> Tuple[int, any]:
    """Performs a GET request towards endpoint

    Args:
        endpoint (str): URL
        params (Dict, optional): GET parameters. Defaults to None.
        err_codes (List[int], optional): List of error codes to handle in calling function. Defaults to None.
        timeout (float, optional): Request timeout in seconds. Defaults to None.

    Returns:
        Tuple[int, any]: Request status code and request data
    """
    return self._do(
        http_method="GET",
        endpoint=endpoint,
        params=params,
        err_codes=err_codes,
        timeout=timeout,
    )

post

1
2
3
4
5
6
7
post(
    endpoint,
    params=None,
    data=None,
    err_codes=None,
    timeout=None,
)

Performs a POST request towards endpoint

Parameters:

Name Type Description Default
endpoint str

URL

required
params Dict

POST parameters. Defaults to None.

None
data Dict

POST data. Defaults to None.

None
err_codes List[int]

List of error codes to handle in calling function. Defaults to None.

None
timeout float

Request timeout in seconds. Defaults to None.

None

Returns:

Type Description
Tuple[int, any]

Tuple[int, any]: Request status code and request data

Source code in fio_wrapper/fio_adapter.py
def post(
    self,
    endpoint: str,
    params: Dict = None,
    data: Dict = None,
    err_codes: Optional[List[int]] = None,
    timeout: Optional[float] = None,
) -> Tuple[int, any]:
    """Performs a POST request towards endpoint

    Args:
        endpoint (str): URL
        params (Dict, optional): POST parameters. Defaults to None.
        data (Dict, optional): POST data. Defaults to None.
        err_codes (List[int], optional): List of error codes to handle in calling function. Defaults to None.
        timeout (float): Request timeout in seconds. Defaults to None.

    Returns:
        Tuple[int, any]: Request status code and request data
    """
    return self._do(
        http_method="POST",
        endpoint=endpoint,
        params=params,
        data=data,
        err_codes=err_codes,
        timeout=timeout,
    )