Base ABC

wingpy.base

RestApiBaseClass

RestApiBaseClass(
    *,
    base_url: str,
    verify: SSLContext | bool = True,
    backoff_initial: int = 1,
    backoff_multiplier: float = 2.0,
    retries: int = 3,
    auth_lifetime: int = 0,
    auth_refresh_percentage: float = 1,
    rate_limit_period: int = 0,
    rate_limit_max_requests: int = 0,
    headers: dict | None = None,
    timeout: int = 10
)

Bases: ABC

An abstract base class for REST API clients.

This class provides a common interface and functionality for interacting with REST APIs. It handles requests, headers, throttling, path parameters, logging, retries, errors and session lifetime. It also defines the abstract methods that must be implemented by clients.

Source code in src/wingpy/base.py

def __init__(
    self,
    *,
    base_url: str,
    verify: SSLContext | bool = True,
    backoff_initial: int = 1,
    backoff_multiplier: float = 2.0,
    retries: int = 3,
    auth_lifetime: int = 0,
    auth_refresh_percentage: float = 1,
    rate_limit_period: int = 0,
    rate_limit_max_requests: int = 0,
    headers: dict | None = None,
    timeout: int = 10,
):
    self._verify_base_url(base_url)
    self.base_url: str = base_url
    """
    A string containing the base URL of the API server.

    The base URL path must include:

     - Scheme (http / https)
     - Hostname / IP address
     - TCP port
     - Base path, if any (MUST NOT end with a `/`)

    Examples
    --------
    - https://api.example.com/api/v1
    - http://api.example.com:8080
    """

    self.auth_lifetime: int = auth_lifetime
    """
    The lifetime in seconds of authentication token.
    """

    self.auth_refresh_percentage: float = auth_refresh_percentage
    """
    The percentage of the authentication token lifetime at which the token should be refreshed.

    This is used to avoid token expiration during long-running requests.
    The value should be between 0 and 1.
    For example, if the token lifetime is 3600 seconds (1 hour) and the refresh percentage is 0.8,
    the token will be refreshed after 2880 seconds (48 minutes).
    """

    self.retries: int = retries
    """
    The number of times a request will be retried in case of failures.

    The first attempt is not counted as a retry.
    """

    self.verify: SSLContext | bool = verify
    """
    Controls the verification of the API server SSL certificate.

    It can simply be enabled or disabled using boolean values,
    or a custom SSLContext can be passed to the constructor to use a custom
    certificate authority.

    Examples
    --------

    - `True`: Verify the server's SSL certificate using the system's CA certificates.
    - `False`: Disable SSL certificate verification.
    - `ssl.create_default_context(cafile="my-custom-ca.pem")`: Use a custom CA certificate for verification.
    """

    self.headers: dict = headers or {}
    """
    A dictionary of HTTP headers to be sent with each request.
    These headers will be merged with any `headers` dict passed to an individual request.
    """

    self.request_index: int = 0
    """
    An index to keep track of the number of requests made.
    """

    self.client: httpx.Client | None = None
    """
    An httpx Client instance used to send requests to the API server.

    This client is created when the first request is made and is reused for all subsequent requests.
    The opened TCP connection is reused for multiple requests to the same server.
    """

    self.auth_timestamp: arrow.Arrow | None = None
    """
    A timestamp indicating when the authentication token was last refreshed.

    In combination with auth_lifetime and auth_refresh_percentage it is used
    to determine when the token should be refreshed again.
    """

    self.path_params: dict = {}
    """
    A dictionary of path parameters to be used in the API path of each request.

    These parameters will be merged with any `path_params` dict passed to the request.
    """

    if self.MAX_CONNECTIONS > 1:
        # Leave one connection for the main thread used for authentication and synchronous requests
        max_workers = self.MAX_CONNECTIONS - 1
    else:
        # If a maximum number of connections is not supported, just use a single worker
        max_workers = 1

    self.tasks: TaskRunner = TaskRunner(max_workers=max_workers)
    """
    Manages concurrent requests to the API server.

    The number of concurrent requests is limited by the MAX_CONNECTIONS property:

    - 1 connection is reserved for the main thread used for authentication and synchronous requests.
    - The remaining connections are used for concurrent requests.

    See Also
    --------
    [`wingpy.scheduling.TaskRunner`](https://wingpy.automation.wingmen.dk/api/scheduling/#wingpy.scheduling.TaskRunner)
        Schedule and run asynchronous tasks in parallel.

    """

    self.throttler: RequestThrottler = RequestThrottler(
        backoff_initial=backoff_initial,
        backoff_multiplier=backoff_multiplier,
        rate_limit_period=rate_limit_period,
        rate_limit_max_requests=rate_limit_max_requests,
    )
    """"
    Manages request throttling and rate limiting to the API server.
    """

    self.request_log: list[RequestLogEntry] = []
    """
    A list of requests made to the API server.

    Each entry contains the request URL, status code, and timestamp.
    """

    self.timeout: int = timeout
    """The timeout in seconds for each request to the API server."""

    self._auth_lock: threading.Lock = threading.Lock()
    """Allow only one thread to authenticate at a time."""

MAX_CONNECTIONS class-attribute

MAX_CONNECTIONS: int

The maximum number of concurrent connections opened to the API server.

If the number is documented in official documentation, it should be used to limit the number of connections. In other cases we may need to limit the number of connections to avoid overwhelming the server or the client machine.

RETRY_RESPONSES class-attribute

RETRY_RESPONSES: list[HttpResponsePattern]

HTTP status codes and response text that should trigger a retry.

Some APIs have specific responses that require a retry, even if the status code is not 429.

auth_lifetime instance-attribute

auth_lifetime: int = auth_lifetime

The lifetime in seconds of authentication token.

auth_refresh_percentage instance-attribute

auth_refresh_percentage: float = auth_refresh_percentage

The percentage of the authentication token lifetime at which the token should be refreshed.

This is used to avoid token expiration during long-running requests. The value should be between 0 and 1. For example, if the token lifetime is 3600 seconds (1 hour) and the refresh percentage is 0.8, the token will be refreshed after 2880 seconds (48 minutes).

auth_timestamp instance-attribute

auth_timestamp: Arrow | None = None

A timestamp indicating when the authentication token was last refreshed.

In combination with auth_lifetime and auth_refresh_percentage it is used to determine when the token should be refreshed again.

base_url instance-attribute

base_url: str = base_url

A string containing the base URL of the API server.

The base URL path must include:

• Scheme (http / https)
• Hostname / IP address
• TCP port
• Base path, if any (MUST NOT end with a /)

Examples:

• https://api.example.com/api/v1
• http://api.example.com:8080

client instance-attribute

client: Client | None = None

An httpx Client instance used to send requests to the API server.

This client is created when the first request is made and is reused for all subsequent requests. The opened TCP connection is reused for multiple requests to the same server.

headers instance-attribute

headers: dict = headers or {}

A dictionary of HTTP headers to be sent with each request. These headers will be merged with any headers dict passed to an individual request.

path_params instance-attribute

path_params: dict = {}

A dictionary of path parameters to be used in the API path of each request.

These parameters will be merged with any path_params dict passed to the request.

request_index instance-attribute

request_index: int = 0

An index to keep track of the number of requests made.

request_log instance-attribute

request_log: list[RequestLogEntry] = []

A list of requests made to the API server.

Each entry contains the request URL, status code, and timestamp.

retries instance-attribute

retries: int = retries

The number of times a request will be retried in case of failures.

The first attempt is not counted as a retry.

tasks instance-attribute

tasks: TaskRunner = TaskRunner(max_workers=max_workers)

Manages concurrent requests to the API server.

The number of concurrent requests is limited by the MAX_CONNECTIONS property:

• 1 connection is reserved for the main thread used for authentication and synchronous requests.
• The remaining connections are used for concurrent requests.

See Also

wingpy.scheduling.TaskRunner Schedule and run asynchronous tasks in parallel.

throttler instance-attribute

throttler: RequestThrottler = RequestThrottler(
    backoff_initial=backoff_initial,
    backoff_multiplier=backoff_multiplier,
    rate_limit_period=rate_limit_period,
    rate_limit_max_requests=rate_limit_max_requests,
)

" Manages request throttling and rate limiting to the API server.

timeout instance-attribute

timeout: int = timeout

The timeout in seconds for each request to the API server.

verify instance-attribute

verify: SSLContext | bool = verify

Controls the verification of the API server SSL certificate.

It can simply be enabled or disabled using boolean values, or a custom SSLContext can be passed to the constructor to use a custom certificate authority.

Examples:

• True: Verify the server's SSL certificate using the system's CA certificates.
• False: Disable SSL certificate verification.
• ssl.create_default_context(cafile="my-custom-ca.pem"): Use a custom CA certificate for verification.

__enter__

__enter__()

When a context manager is used, this method is called to enter the runtime context. This method is called when the with statement is used with this class.

Returns:

Type	Description
self	The instance of the class itself.

Source code in src/wingpy/base.py

def __enter__(self):
    """
    When a context manager is used, this method is called to enter the runtime context.
    This method is called when the `with` statement is used with this class.

    Returns
    -------
    self
        The instance of the class itself.
    """
    return self

__exit__

__exit__(exc_type, exc_value, traceback) -> bool

This method is called when exiting the runtime context. It closes the client and handles any exceptions that occurred during the context.

Parameters:

Name	Type	Description	Default
exc_type		The type of the exception that occurred, if any.	required
exc_value		The value of the exception that occurred, if any.	required
traceback		The traceback object of the exception that occurred, if any.	required

Returns:

Type	Description
bool	False to propagate the exception, True to suppress it. Always returns False. This ensures that any exceptions that occurred during the context are propagated.

Source code in src/wingpy/base.py

def __exit__(self, exc_type, exc_value, traceback) -> bool:
    """
    This method is called when exiting the runtime context.
    It closes the client and handles any exceptions that occurred during the context.

    Parameters
    ----------
    exc_type
        The type of the exception that occurred, if any.
    exc_value
        The value of the exception that occurred, if any.
    traceback
        The traceback object of the exception that occurred, if any.

    Returns
    -------
    bool
        `False` to propagate the exception, `True` to suppress it. Always returns `False`.

        This ensures that any exceptions that occurred during the context are propagated.
    """
    self.close()
    if exc_type is not None:
        logger.error(f"Exception occurred: {exc_value}")
    # Return False to propagate the exception
    return False

authenticate

authenticate() -> None

Executes the API-specific authentication process and records timestamps for session tracking.

Notes

Authentication will automatically be carried out just-in-time.

Only call this method directly if you need to authenticate proactively, outside of normal request flow.

Source code in src/wingpy/base.py

def authenticate(self) -> None:
    """
    Executes the API-specific authentication process and records timestamps
    for session tracking.

    Notes
    ----
    Authentication will automatically be carried out just-in-time.

    Only call this method directly if you need to authenticate proactively,
    outside of normal request flow.
    """

    # Authenticate
    logger.debug("Authenticating and recording token lifetime.")
    auth_response = self._authenticate()

    # Record the time of authentication
    self.auth_timestamp = arrow.utcnow()

    self._after_auth(auth_response=auth_response)

build_url

build_url(path: str, path_params: dict)

Constructs the full URL for a request.

Combines the base URL with the provided path and substituting any path parameters. Path parameters are variables embedded into the URL path using {} braces. Example: /organizations/{organizationId}/firmware/upgrades Example: /api/fmc_config/v1/domain/{domainUUID}/policy/accesspolicies/{objectId} Reusable path parameters can be added to the class instance using the path_params attribute. Single-use path parameters can be passed as a dictionary to the path_params argument.

Parameters:

Name	Type	Description	Default
path	str	The URL path, which may include placeholders for path parameters (e.g., "/organizations/{organizationId}/firmware/upgrades").	required
path_params	dict	A dictionary of path parameters to be used in the API path. Is merged with self.path_params.	required

Returns:

Type	Description
str	The fully constructed URL with all path parameters substituted.

Raises:

Type	Description
InvalidEndpointError	If the resulting URL contains unsubstituted path parameters.

Source code in src/wingpy/base.py

def build_url(self, path: str, path_params: dict):
    """
    Constructs the full URL for a request.

    Combines the base URL with the provided path and substituting any path parameters.
    Path parameters are variables embedded into the URL path using {} braces.
    __Example:__ /organizations/{organizationId}/firmware/upgrades
    __Example:__ /api/fmc_config/v1/domain/{domainUUID}/policy/accesspolicies/{objectId}
    Reusable path parameters can be added to the class instance using the `path_params` attribute.
    Single-use path parameters can be passed as a dictionary to the `path_params` argument.

    Parameters
    ----------
    path
        The URL path, which may include placeholders for path parameters
        (e.g., "/organizations/{organizationId}/firmware/upgrades").

    path_params
        A dictionary of path parameters to be used in the API path.
        Is merged with [`self.path_params`](https://wingpy.automation.wingmen.dk/api/base/#wingpy.base.RestApiBaseClass.path_params).

    Returns
    -------
    str
        The fully constructed URL with all path parameters substituted.

    Raises
    ------

    InvalidEndpointError
        If the resulting URL contains unsubstituted path parameters.
    """

    # Merge reusable and single-use path parameters
    merged_path_params = self.path_params.copy()
    merged_path_params.update(path_params or {})

    # Replace path parameters in the URL and report any missing parameters
    # that are not in the path_params dictionary
    try:
        url = f"{self.base_url}{path}".format(**merged_path_params)
    except KeyError as e:
        raise InvalidEndpointError(
            f"Missing path parameter: {e}. Available parameters: {', '.join(merged_path_params.keys())}"
        )

    return url

close

close() -> None

Close the httpx client and release any resources. This method should be called when the client is no longer needed. It is automatically called when exiting the context manager.

Source code in src/wingpy/base.py

def close(self) -> None:
    """
    Close the httpx client and release any resources.
    This method should be called when the client is no longer needed.
    It is automatically called when exiting the context manager.
    """
    if self.client:
        self.client.close()
        logger.debug("Closed API client")
        self.client = None
        logger.debug("Client set to None")

delete abstractmethod

delete(path: str, **kwargs) -> httpx.Response

Abstract method to send a DELETE request to the API server.

Applies any API-specific pre- or post-processing.

Source code in src/wingpy/base.py

@abstractmethod
def delete(self, path: str, **kwargs) -> httpx.Response:
    """
    Abstract method to send a DELETE request to the API server.

    Applies any API-specific pre- or post-processing.
    """
    raise NotImplementedError

get abstractmethod

get(path: str, **kwargs) -> httpx.Response

Abstract method to send a GET request to the API server.

Applies any API-specific pre- or post-processing.

Source code in src/wingpy/base.py

@abstractmethod
def get(self, path: str, **kwargs) -> httpx.Response:
    """
    Abstract method to send a GET request to the API server.

    Applies any API-specific pre- or post-processing.
    """
    raise NotImplementedError

is_authenticated abstractmethod

is_authenticated() -> bool

Abstract method to check if the client is authenticated with the API server.

Returns:

Type	Description
bool	True if authenticated, False otherwise.

Source code in src/wingpy/base.py

@abstractmethod
def is_authenticated(self) -> bool:
    """
    Abstract method to check if the client is authenticated with the API server.

    Returns
    -------
    bool
        `True` if authenticated, `False` otherwise.
    """
    raise NotImplementedError

is_retry_response

is_retry_response(response: Response, method: str) -> bool

Source code in src/wingpy/base.py

def is_retry_response(self, response: httpx.Response, method: str) -> bool:
    result = False
    for retry_reponse in self.RETRY_RESPONSES:
        if (
            response.status_code in retry_reponse.status_codes
            and method in retry_reponse.methods
        ):
            for content_pattern in retry_reponse.content_patterns:
                if content_pattern.match(response.content.decode()):
                    result = True
                    break
        if result:
            break
    return result

patch abstractmethod

patch(path: str, **kwargs) -> httpx.Response

Abstract method to send a PATCH request to the API server.

Applies any API-specific pre- or post-processing.

Source code in src/wingpy/base.py

@abstractmethod
def patch(self, path: str, **kwargs) -> httpx.Response:
    """
    Abstract method to send a PATCH request to the API server.

    Applies any API-specific pre- or post-processing.
    """
    raise NotImplementedError

post abstractmethod

post(path: str, **kwargs) -> httpx.Response

Abstract method to send a POST request to the API server.

Applies any API-specific pre- or post-processing.

Source code in src/wingpy/base.py

@abstractmethod
def post(self, path: str, **kwargs) -> httpx.Response:
    """
    Abstract method to send a POST request to the API server.

    Applies any API-specific pre- or post-processing.
    """
    raise NotImplementedError

put abstractmethod

put(path: str, **kwargs) -> httpx.Response

Abstract method to send a PUT request to the API server.

Applies any API-specific pre- or post-processing.

Source code in src/wingpy/base.py

@abstractmethod
def put(self, path: str, **kwargs) -> httpx.Response:
    """
    Abstract method to send a PUT request to the API server.

    Applies any API-specific pre- or post-processing.
    """
    raise NotImplementedError

request

request(
    method: str,
    path: str,
    data: dict | list | _Element | str | None,
    params: dict,
    path_params: dict,
    headers: dict,
    timeout: int,
    is_auth_endpoint: bool,
    auth: Auth | None,
) -> httpx.Response

Send any type of HTTP request and receive response.

Handles any preprocessing of authentication, parameters, payload, and URL

Parameters:

Name	Type	Description	Default
method	str	The HTTP method to use for the request (GET, POST, PUT, PATCH, DELETE).	required
path	str	URL endpoint path. Is combined with self.base_url	required
params	dict	Query parameters to include in the request.	required
path_params	dict	A dictionary of path parameters to be used in the API path. Is merged with self.path_params	required
headers	dict	A dictionary of HTTP headers to be sent with the request. Is merged with self.headers	required
timeout	int	Number of seconds to wait for HTTP responses before raising httpx.TimeoutException exception.	required
is_auth_endpoint	bool	A boolean indicating if the request is a dedicated authentication request. If True, authentication flow is skipped.	required
auth	Auth | None	The authentication object to be used for the request.	required

Returns:

Type	Description
Response	The response object returned by the API server.

Source code in src/wingpy/base.py

def request(
    self,
    method: str,
    path: str,
    data: dict | list | etree._Element | str | None,
    params: dict,
    path_params: dict,
    headers: dict,
    timeout: int,
    is_auth_endpoint: bool,
    auth: httpx.Auth | None,
) -> httpx.Response:
    """
    Send any type of HTTP request and receive response.

    Handles any preprocessing of authentication, parameters, payload, and URL

    Parameters
    ----------
    method
        The HTTP method to use for the request (`GET`, `POST`, `PUT`, `PATCH`, `DELETE`).

    path : str
        URL endpoint path. Is combined with [`self.base_url`](https://wingpy.automation.wingmen.dk/api/base/#wingpy.base.RestApiBaseClass.base_url)

    params : dict
        Query parameters to include in the request.

    path_params
        A dictionary of path parameters to be used in the API path.
        Is merged with [`self.path_params`](https://wingpy.automation.wingmen.dk/api/base/#wingpy.base.RestApiBaseClass.path_params)

    headers
        A dictionary of HTTP headers to be sent with the request.
        Is merged with [`self.headers`](https://wingpy.automation.wingmen.dk/api/base/#wingpy.base.RestApiBaseClass.headers)

    timeout : int
        Number of seconds to wait for HTTP responses before raising httpx.TimeoutException exception.

    is_auth_endpoint : bool
        A boolean indicating if the request is a dedicated authentication request.

        If `True`, authentication flow is skipped.

    auth : httpx.Auth | None
        The authentication object to be used for the request.

    Returns
    -------
    httpx.Response
        The response object returned by the API server.
    """

    # If the request is not an authentication request, make sure we have a valid token
    self._ensure_client()

    if not is_auth_endpoint:
        self._ensure_auth()

    request = self._prepare_request(
        method,
        path,
        params=params,
        path_params=path_params,
        headers=headers,
        timeout=timeout,
        data=data,
    )

    return self._send_request_with_retry(
        request, is_auth_endpoint=is_auth_endpoint, auth=auth
    )

serialize_payload

serialize_payload(
    *, data: dict | list | _Element | str | None
) -> str

Source code in src/wingpy/base.py

def serialize_payload(
    self, *, data: dict | list | etree._Element | str | None
) -> str:
    if isinstance(data, (dict, list)):
        json_data = json.dumps(data)
        return json_data
    elif isinstance(data, etree._Element):
        xml_data = etree.tostring(data)
        return xml_data
    elif isinstance(data, str):
        return data
    elif data is None:
        return ""
    else:
        raise ValueError(
            "Data for request payload must be provided as string, dict, list or lxml.etree.Element."
        )