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.

OpenAPI

OpenAPI(
    base_url: str,
    doc_path: str,
    headers: Optional[Dict[str, str]] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    cert: Optional[str] = None,
    key: Optional[str] = None,
    validate_certs: bool = True,
    refresh_cache: bool = False,
    safe_calls_only: bool = False,
    debug_callback: Optional[
        Callable[[int, str], Any]
    ] = None,
    user_agent: Optional[str] = None,
    cid: 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.

  • username (Optional[str], default: None ) –

    Username used for basic auth.

  • password (Optional[str], default: None ) –

    Password used for basic auth.

  • 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
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def __init__(
    self,
    base_url: str,
    doc_path: str,
    headers: Optional[Dict[str, str]] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    cert: Optional[str] = None,
    key: Optional[str] = None,
    validate_certs: bool = True,
    refresh_cache: bool = False,
    safe_calls_only: bool = False,
    debug_callback: Optional[Callable[[int, str], Any]] = None,
    user_agent: Optional[str] = None,
    cid: Optional[str] = None,
):
    if not validate_certs:
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    self.debug_callback: Callable[[int, str], 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 username and password:
        if cert or key:
            raise OpenAPIError(_("Cannot use both username/password and cert auth."))
        self._session.auth = (username, password)
    elif username:
        raise OpenAPIError(_("Password is required if username is set."))
    elif password:
        raise OpenAPIError(_("Username is required if password is set."))
    elif 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: Optional[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: Optional[Dict[str, Any]] = None,
    body: Optional[Dict[str, Any]] = None,
    validate_body: bool = True,
) -> 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
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
def call(
    self,
    operation_id: str,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[Dict[str, Any]] = None,
    validate_body: bool = True,
) -> 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)