pulp_glue.common.openapi¶

OpenAPIError ¶

Bases: Exception

Base Exception for errors related to using the openapi spec.

OpenAPIValidationError ¶

Bases: OpenAPIError

Exception raised for failed client side validation of parameters or request bodies.

UnsafeCallError ¶

Bases: OpenAPIError

Exception raised for POST, PUT, PATCH or DELETE calls with safe_calls_only=True.

AuthProviderBase ¶

Base class for auth providers.

This abstract base class will analyze the authentication proposals of the openapi specs. Different authentication schemes should be implemented by subclasses. Returned auth objects need to be compatible with requests.auth.AuthBase.

basic_auth ¶

basic_auth() -> (
    t.Optional[
        t.Union[t.Tuple[str, str], requests.auth.AuthBase]
    ]
)

Implement this to provide means of http basic auth.

Source code in pulp-glue/pulp_glue/common/openapi.py

def basic_auth(self) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.AuthBase]]:
    """Implement this to provide means of http basic auth."""
    return None

BasicAuthProvider ¶

BasicAuthProvider(username: str, password: str)

Bases: AuthProviderBase

Reference Implementation for AuthProviderBase providing basic auth with username, password.

Source code in pulp-glue/pulp_glue/common/openapi.py

def __init__(self, username: str, password: str):
    self.username = username
    self.password = password

OpenAPI ¶

OpenAPI(
    base_url: str,
    doc_path: str,
    headers: t.Optional[t.Dict[str, str]] = None,
    auth_provider: t.Optional[AuthProviderBase] = None,
    cert: t.Optional[str] = None,
    key: t.Optional[str] = None,
    validate_certs: bool = True,
    refresh_cache: bool = False,
    safe_calls_only: bool = False,
    debug_callback: t.Optional[
        t.Callable[[int, str], t.Any]
    ] = None,
    user_agent: t.Optional[str] = None,
    cid: t.Optional[str] = None,
)

The abstraction Layer to interact with a server providing an openapi v3 specification.

Parameters:

base_url (str) –

The base URL inlcuding the HTTP scheme, hostname and optional subpaths of the served api.
doc_path (str) –

Path of the json api doc schema relative to the base_url.
headers (Optional[Dict[str, str]], default: None ) –

Dictionary of additional request headers.
auth_provider (Optional[AuthProviderBase], default: None ) –

Object that returns requests auth objects according to the api spec.
cert (Optional[str], default: None ) –

Client certificate used for auth.
key (Optional[str], default: None ) –

Matching key for cert if not already included.
validate_certs (bool, default: True ) –

Whether to check server TLS certificates agains a CA.
refresh_cache (bool, default: False ) –

Whether to fetch the api doc regardless.
safe_calls_only (bool, default: False ) –

Flag to disallow issuing POST, PUT, PATCH or DELETE calls.
debug_callback (Optional[Callable[[int, str], Any]], default: None ) –

Callback that will be called with strings useful for logging or debugging.
user_agent (Optional[str], default: None ) –

String to use in the User-Agent header.
cid (Optional[str], default: None ) –

Correlation ID to send with all requests.

Source code in pulp-glue/pulp_glue/common/openapi.py

def __init__(
    self,
    base_url: str,
    doc_path: str,
    headers: t.Optional[t.Dict[str, str]] = None,
    auth_provider: t.Optional[AuthProviderBase] = None,
    cert: t.Optional[str] = None,
    key: t.Optional[str] = None,
    validate_certs: bool = True,
    refresh_cache: bool = False,
    safe_calls_only: bool = False,
    debug_callback: t.Optional[t.Callable[[int, str], t.Any]] = None,
    user_agent: t.Optional[str] = None,
    cid: t.Optional[str] = None,
):
    if not validate_certs:
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    self.debug_callback: t.Callable[[int, str], t.Any] = debug_callback or (lambda i, x: None)
    self.base_url: str = base_url
    self.doc_path: str = doc_path
    self.safe_calls_only: bool = safe_calls_only

    self._session: requests.Session = requests.session()
    if auth_provider:
        if cert or key:
            raise OpenAPIError(_("Cannot use both 'auth' and 'cert'."))
        self.auth_provider = auth_provider
    else:
        if cert and key:
            self._session.cert = (cert, key)
        elif cert:
            self._session.cert = cert
        elif key:
            raise OpenAPIError(_("Cert is required if key is set."))
    self._session.headers.update(
        {
            "User-Agent": user_agent or f"Pulp-glue openapi parser ({__version__})",
            "Accept": "application/json",
        }
    )
    if cid:
        self._session.headers["Correlation-Id"] = cid
    if headers:
        self._session.headers.update(headers)
    self._session.max_redirects = 0

    verify: t.Optional[t.Union[bool, str]] = (
        os.environ.get("PULP_CA_BUNDLE") if validate_certs is not False else False
    )
    session_settings = self._session.merge_environment_settings(
        base_url, {}, None, verify, None
    )
    self._session.verify = session_settings["verify"]
    self._session.proxies = session_settings["proxies"]

    self.load_api(refresh_cache=refresh_cache)

call ¶

call(
    operation_id: str,
    parameters: t.Optional[t.Dict[str, t.Any]] = None,
    body: t.Optional[t.Dict[str, t.Any]] = None,
    validate_body: bool = True,
) -> t.Any

Make a call to the server.

Parameters:

operation_id (str) –

ID of the operation in the openapi v3 specification.
parameters (Optional[Dict[str, Any]], default: None ) –

Arguments that are to be sent as headers, querystrings or part of the URI.
body (Optional[Dict[str, Any]], default: None ) –

Body payload for POST, PUT, PATCH calls.
validate_body (bool, default: True ) –

Indicate whether the body should be validated.

Returns:	`Any` – The JSON decoded server response if any.

Raises:	`OpenAPIValidationError` – on failed input validation (no request was sent to the server). `HTTPError` – on failures related to the HTTP call made.

Source code in pulp-glue/pulp_glue/common/openapi.py

def call(
    self,
    operation_id: str,
    parameters: t.Optional[t.Dict[str, t.Any]] = None,
    body: t.Optional[t.Dict[str, t.Any]] = None,
    validate_body: bool = True,
) -> t.Any:
    """
    Make a call to the server.

    Parameters:
        operation_id: ID of the operation in the openapi v3 specification.
        parameters: Arguments that are to be sent as headers, querystrings or part of the URI.
        body: Body payload for POST, PUT, PATCH calls.
        validate_body: Indicate whether the body should be validated.

    Returns:
        The JSON decoded server response if any.

    Raises:
        OpenAPIValidationError: on failed input validation (no request was sent to the server).
        requests.HTTPError: on failures related to the HTTP call made.
    """
    method, path = self.operations[operation_id]
    path_spec = self.api_spec["paths"][path]
    method_spec = path_spec[method]

    if parameters is None:
        parameters = {}
    else:
        parameters = parameters.copy()

    if any(self.extract_params("cookie", path_spec, method_spec, parameters)):
        raise NotImplementedError(_("Cookie parameters are not implemented."))

    headers = self.extract_params("header", path_spec, method_spec, parameters)

    for name, value in self.extract_params("path", path_spec, method_spec, parameters).items():
        path = path.replace("{" + name + "}", value)

    query_params = self.extract_params("query", path_spec, method_spec, parameters)

    if any(parameters):
        raise OpenAPIError(
            _("Parameter [{names}] not available for {operation_id}.").format(
                names=", ".join(parameters.keys()), operation_id=operation_id
            )
        )
    url = urljoin(self.base_url, path)

    request: requests.PreparedRequest = self.render_request(
        path_spec,
        method,
        url,
        query_params,
        headers,
        body,
        validate_body=validate_body,
    )

    self.debug_callback(1, f"{operation_id} : {method} {request.url}")
    for key, value in request.headers.items():
        self.debug_callback(2, f"  {key}: {value}")
    if request.body is not None:
        self.debug_callback(3, f"{request.body!r}")
    if self.safe_calls_only and method.upper() not in SAFE_METHODS:
        raise UnsafeCallError(_("Call aborted due to safe mode"))
    try:
        response: requests.Response = self._session.send(request)
    except requests.TooManyRedirects as e:
        assert e.response is not None
        raise OpenAPIError(
            _("Received redirect to '{url}'. Please check your CLI configuration.").format(
                url=e.response.headers["location"]
            )
        )
    except requests.RequestException as e:
        raise OpenAPIError(str(e))
    self.debug_callback(
        1, _("Response: {status_code}").format(status_code=response.status_code)
    )
    for key, value in response.headers.items():
        self.debug_callback(2, f"  {key}: {value}")
    if response.text:
        self.debug_callback(3, f"{response.text}")
    if "Correlation-ID" in response.headers:
        self._set_correlation_id(response.headers["Correlation-ID"])
    response.raise_for_status()
    return self.parse_response(method_spec, response)