oapi.client

SSLContext

SSLContext(check_hostname: bool = True)

Bases: ssl.SSLContext

This class is a wrapper for ssl.SSLContext which makes it possible to connect to hosts which have an unverified SSL certificate.

Source code in src/oapi/client.py

def __init__(
    self,
    check_hostname: bool = True,  # noqa: FBT001 FBT002
) -> None:
    if check_hostname:
        self.load_default_certs()
    else:
        self.check_hostname: bool = False
        self.verify_mode: ssl.VerifyMode = ssl.CERT_NONE
    super().__init__()

__reduce__

__reduce__() -> tuple

A pickled instance of this class will just be an entirely new instance.

Source code in src/oapi/client.py

def __reduce__(self) -> tuple:
    """
    A pickled instance of this class will just be an entirely new
    instance.
    """
    return SSLContext, (self.check_hostname,)

Client

Client(
    url: str | None = None,
    *,
    user: str | None = None,
    password: str | None = None,
    bearer_token: str | None = None,
    api_key: str | None = None,
    api_key_in: typing.Literal[
        "header", "query", "cookie"
    ] = "header",
    api_key_name: str = "X-API-KEY",
    oauth2_client_id: str | None = None,
    oauth2_client_secret: str | None = None,
    oauth2_username: str | None = None,
    oauth2_password: str | None = None,
    oauth2_authorization_url: str | None = None,
    oauth2_token_url: str | None = None,
    oauth2_scope: str | tuple[str, ...] | None = None,
    oauth2_refresh_url: str | None = None,
    oauth2_flows: (
        tuple[
            typing.Literal[
                "authorizationCode",
                "implicit",
                "password",
                "clientCredentials",
                "accessCode",
                "application",
            ],
            ...,
        ]
        | None
    ) = None,
    open_id_connect_url: str | None = None,
    headers: (
        collections.abc.Mapping[str, str]
        | collections.abc.Sequence[tuple[str, str]]
    ) = (
        ("Accept", "application/json"),
        ("Content-type", "application/json"),
    ),
    timeout: int = 0,
    retry_number_of_attempts: int = 1,
    retry_for_errors: tuple[
        type[Exception], ...
    ] = oapi.client.DEFAULT_RETRY_FOR_ERRORS,
    retry_hook: typing.Callable[
        [Exception], bool
    ] = oapi.client.default_retry_hook,
    verify_ssl_certificate: bool = True,
    logger: logging.Logger | None = None,
    echo: bool = False
)

A base class for OpenAPI clients.

Parameters:

• url (str | None, default: None ) –

  The base URL for API requests.

• user (str | None, default: None ) –

  A user name for use with HTTP basic authentication.

• password (str | None, default: None ) –

  A password for use with HTTP basic authentication.

• bearer_token (str | None, default: None ) –

  A token for use with HTTP bearer authentication.

• api_key (str | None, default: None ) –

  An API key with which to authenticate requests.

• api_key_in (typing.Literal['header', 'query', 'cookie'], default: 'header' ) –

  Where the API key should be conveyed: "header", "query" or "cookie".

• api_key_name (str, default: 'X-API-KEY' ) –

  The name of the header, query parameter, or cookie parameter in which to convey the API key.

• oauth2_client_id (str | None, default: None ) –

  An OAuth2 client ID.

• oauth2_client_secret (str | None, default: None ) –

  An OAuth2 client secret.

• oauth2_username (str | None, default: None ) –

  A username for the "password" OAuth2 grant type.

• oauth2_password (str | None, default: None ) –

  A password for the "password" OAuth2 grant type.

• oauth2_authorization_url (str | None, default: None ) –

  The authorization URL to use for an OAuth2 flow. Can be relative to url.

• oauth2_token_url (str | None, default: None ) –

  The token URL to use for OAuth2 authentication. Can be relative to url.

• oauth2_refresh_url (str | None, default: None ) –

  The URL to be used for obtaining refresh tokens for OAuth2 authentication.

• oauth2_flows (tuple[typing.Literal['authorizationCode', 'implicit', 'password', 'clientCredentials', 'accessCode', 'application'], ...] | None, default: None ) –

  A tuple containing one or more of the following: "authorizationCode", "implicit", "password" and/or "clientCredentials".

• open_id_connect_url (str | None, default: None ) –

  An OpenID connect URL where a JSON web token containing OAuth2 information can be found.

• headers (collections.abc.Mapping[str, str] | collections.abc.Sequence[tuple[str, str]], default: (('Accept', 'application/json'), ('Content-type', 'application/json')) ) –

  Default headers to include with all requests. Method-specific header arguments will override or modify these, where applicable, as will dynamically modified headers such as content-length, authorization, cookie, etc.

• timeout (int, default: 0 ) –

  The number of seconds before a request will timeout and throw an error. If this is 0 (the default), the system default timeout will be used.

• retry_number_of_attempts (int, default: 1 ) –

  The number of times to retry a request which results in an error.

• retry_for_errors (tuple[type[Exception], ...], default: oapi.client.DEFAULT_RETRY_FOR_ERRORS ) –

  A tuple of one or more exception types on which to retry a request. To retry for all errors, pass (Exception,) for this argument.

• retry_hook (typing.Callable[[Exception], bool], default: oapi.client.default_retry_hook ) –

  A function, accepting one argument (an Exception), and returning a boolean value indicating whether to retry the request (if retries have not been exhausted). This hook applies only for exceptions which are a sub-class of an exception included in retry_for_errors.

• verify_ssl_certificate (bool, default: True ) –

  If True, SSL certificates are verified, per usual. If False, SSL certificates are not verified.

• logger (logging.Logger | None, default: None ) –

  A logging.Logger to which requests should be logged.

• echo (bool, default: False ) –

  If True, requests/responses are printed as they occur.

Source code in src/oapi/client.py

def __init__(
    self,
    url: str | None = None,
    *,
    user: str | None = None,
    password: str | None = None,
    bearer_token: str | None = None,
    api_key: str | None = None,
    api_key_in: typing.Literal["header", "query", "cookie"] = "header",
    api_key_name: str = "X-API-KEY",
    oauth2_client_id: str | None = None,
    oauth2_client_secret: str | None = None,
    oauth2_username: str | None = None,
    oauth2_password: str | None = None,
    oauth2_authorization_url: str | None = None,
    oauth2_token_url: str | None = None,
    oauth2_scope: str | tuple[str, ...] | None = None,
    oauth2_refresh_url: str | None = None,
    oauth2_flows: tuple[
        typing.Literal[
            "authorizationCode",
            "implicit",
            "password",
            "clientCredentials",
            # OpenAPI 2.x Compatibility:
            "accessCode",
            "application",
        ],
        ...,
    ]
    | None = None,
    open_id_connect_url: str | None = None,
    headers: collections.abc.Mapping[str, str]
    | collections.abc.Sequence[tuple[str, str]] = (
        ("Accept", "application/json"),
        ("Content-type", "application/json"),
    ),
    timeout: int = 0,
    retry_number_of_attempts: int = 1,
    retry_for_errors: tuple[
        type[Exception], ...
    ] = DEFAULT_RETRY_FOR_ERRORS,
    retry_hook: typing.Callable[  # Force line-break retention
        [Exception], bool
    ] = default_retry_hook,
    verify_ssl_certificate: bool = True,
    logger: Logger | None = None,
    echo: bool = False,
) -> None:
    """
    Parameters:
        url: The base URL for API requests.
        user: A user name for use with HTTP basic authentication.
        password:  A password for use with HTTP basic authentication.
        bearer_token: A token for use with HTTP bearer authentication.
        api_key: An API key with which to authenticate requests.
        api_key_in: Where the API key should be conveyed:
            "header", "query" or "cookie".
        api_key_name: The name of the header, query parameter, or
            cookie parameter in which to convey the API key.
        oauth2_client_id: An OAuth2 client ID.
        oauth2_client_secret: An OAuth2 client secret.
        oauth2_username: A *username* for the "password" OAuth2 grant
            type.
        oauth2_password: A *password* for the "password" OAuth2 grant
            type.
        oauth2_authorization_url: The authorization URL to use for an
            OAuth2 flow. Can be relative to `url`.
        oauth2_token_url: The token URL to use for OAuth2
            authentication. Can be relative to `url`.
        oauth2_refresh_url: The URL to be used for obtaining refresh
            tokens for OAuth2 authentication.
        oauth2_flows: A tuple containing one or more of the
            following: "authorizationCode", "implicit", "password" and/or
            "clientCredentials".
        open_id_connect_url: An OpenID connect URL where a JSON
            web token containing OAuth2 information can be found.
        headers: Default headers to include with all requests.
            Method-specific header arguments will override or modify these,
            where applicable, as will dynamically modified headers such as
            content-length, authorization, cookie, etc.
        timeout: The number of seconds before a request will timeout
            and throw an error. If this is 0 (the default), the system
            default timeout will be used.
        retry_number_of_attempts: The number of times to retry
            a request which results in an error.
        retry_for_errors: A tuple of one or more exception types
            on which to retry a request. To retry for *all* errors,
            pass `(Exception,)` for this argument.
        retry_hook: A function, accepting one argument (an Exception),
            and returning a boolean value indicating whether to retry the
            request (if retries have not been exhausted). This hook applies
            *only* for exceptions which are a sub-class of an exception
            included in `retry_for_errors`.
        verify_ssl_certificate: If `True`, SSL certificates
            are verified, per usual. If `False`, SSL certificates are *not*
            verified.
        logger:
            A `logging.Logger` to which requests should be logged.
        echo: If `True`, requests/responses are printed as
            they occur.
    """
    message: str
    # Ensure the API key location is valid
    if api_key_in not in ("header", "query", "cookie"):
        message = (
            f"Invalid input for `api_key_in`:  {api_key_in!r}\n"
            'Valid values are "header", "query" or "cookie".'
        )
        raise ValueError(message)
    # Translate OpenAPI 2x OAuth2 flow names
    if oauth2_flows:
        flow: str
        oauth2_flows = tuple(
            (
                "authorizationCode"
                if flow == "accessCode"
                else "clientCredentials"
                if flow == "application"
                else flow
            )
            for flow in oauth2_flows
        )
    # Ensure OAuth2 flows are valid
    if oauth2_flows and not all(
        flow
        in {
            "authorizationCode",
            "implicit",
            "password",
            "clientCredentials",
        }
        for flow in oauth2_flows
    ):
        message = (
            f"Invalid value(s) for `oauth2_flows`:  {api_key_in!r}\n"
            'Valid values are "authorizationCode", "implicit", '
            '"password", or "clientCredentials".'
        )
        raise ValueError(message)
    # Ensure URLs are HTTP/HTTPS (security concern) *or* are relative URLs
    url_: str | None
    for url_ in (
        url,
        oauth2_authorization_url,
        oauth2_token_url,
        oauth2_refresh_url,
        open_id_connect_url,
    ):
        if url_ and (
            url_.lower().rpartition(":")[0] not in ("http", "https", "")
        ):
            raise ValueError(url_)
    # Set properties
    self.url: str | None = url
    self.user: str | None = user
    self.password: str | None = password
    self.bearer_token: str | None = bearer_token
    self.api_key: str | None = api_key
    self.api_key_in: typing.Literal["header", "query", "cookie"] = (
        api_key_in
    )
    self.api_key_name: str = api_key_name
    self.oauth2_client_id: str | None = oauth2_client_id
    self.oauth2_client_secret: str | None = oauth2_client_secret
    self.oauth2_username: str | None = oauth2_username
    self.oauth2_password: str | None = oauth2_password
    self.oauth2_authorization_url: str | None = oauth2_authorization_url
    self.oauth2_token_url: str | None = oauth2_token_url
    self.oauth2_scope: str | tuple[str, ...] | None = oauth2_scope
    self.oauth2_refresh_url: str | None = oauth2_refresh_url
    self.oauth2_flows: (
        typing.Literal[
            "authorizationCode",
            "implicit",
            "password",
            "clientCredentials",
        ]
        | None
    ) = oauth2_flows  # type: ignore
    self.open_id_connect_url: str | None = open_id_connect_url
    self.headers: dict[str, str] = dict(headers)
    self.timeout: int = timeout
    self.retry_number_of_attempts: int = retry_number_of_attempts
    self.retry_for_errors: tuple[type[Exception], ...] = retry_for_errors
    self.retry_hook: typing.Callable[[Exception], bool] = retry_hook
    self.verify_ssl_certificate: bool = verify_ssl_certificate
    self.logger: Logger | None = logger
    self.echo: bool = echo
    # Support for persisting cookies
    self._cookie_jar: CookieJar = CookieJar()
    self.__opener: OpenerDirector | None = None
    self._oauth2_authorization_expires: int = 0

request

request(
    path: str,
    method: str,
    *,
    json: str | bytes | sob.abc.Model | None = None,
    data: (
        collections.abc.Mapping[
            str, sob.abc.MarshallableTypes
        ]
        | collections.abc.Sequence[
            tuple[str, sob.abc.MarshallableTypes]
        ]
    ) = (),
    query: (
        collections.abc.Mapping[
            str, sob.abc.MarshallableTypes
        ]
        | collections.abc.Sequence[
            tuple[str, sob.abc.MarshallableTypes]
        ]
        | str
    ) = (),
    headers: (
        collections.abc.Mapping[
            str, sob.abc.MarshallableTypes
        ]
        | collections.abc.Sequence[
            tuple[str, sob.abc.MarshallableTypes]
        ]
    ) = (),
    multipart: bool = False,
    multipart_data_headers: (
        collections.abc.Mapping[
            str, collections.abc.MutableMapping[str, str]
        ]
        | collections.abc.Sequence[
            tuple[
                str,
                collections.abc.MutableMapping[str, str],
            ]
        ]
    ) = (),
    timeout: int = 0
) -> sob.abc.Readable

Construct and submit an HTTP request and return the response (an instance of http.client.HTTPResponse).

Parameters:

• path (str) –

  This is the path of the request, relative to the server base URL

• query (collections.abc.Mapping[str, sob.abc.MarshallableTypes] | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]] | str, default: () ) –
• json (str | bytes | sob.abc.Model | None, default: None ) –

  JSON data to be conveyed in the body of the request

• data (collections.abc.Mapping[str, sob.abc.MarshallableTypes] | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]], default: () ) –

  Form data to be conveyed in the body of the request

• multipart (bool, default: False ) –

  If True, data should be conveyed as a multipart request.

• query (collections.abc.Mapping[str, sob.abc.MarshallableTypes] | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]] | str, default: () ) –

  A dictionary from which to assemble the query string.

• headers (collections.abc.Mapping[str, sob.abc.MarshallableTypes] | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]], default: () ) –
• multipart_data_headers (collections.abc.Mapping[str, collections.abc.MutableMapping[str, str]] | collections.abc.Sequence[tuple[str, collections.abc.MutableMapping[str, str]]], default: () ) –
• timeout (int, default: 0 ) –

Source code in src/oapi/client.py

def request(
    self,
    path: str,
    method: str,
    *,
    json: str | bytes | sob.abc.Model | None = None,
    data: collections.abc.Mapping[str, sob.abc.MarshallableTypes]
    | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]] = (),
    query: collections.abc.Mapping[str, sob.abc.MarshallableTypes]
    | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]]
    | str = (),
    headers: collections.abc.Mapping[str, sob.abc.MarshallableTypes]
    | collections.abc.Sequence[tuple[str, sob.abc.MarshallableTypes]] = (),
    multipart: bool = False,
    multipart_data_headers: collections.abc.Mapping[
        str, collections.abc.MutableMapping[str, str]
    ]
    | collections.abc.Sequence[
        tuple[str, collections.abc.MutableMapping[str, str]]
    ] = (),
    timeout: int = 0,
) -> sob.abc.Readable:
    """
    Construct and submit an HTTP request and return the response
    (an instance of `http.client.HTTPResponse`).

    Parameters:
        path: This is the path of the request, relative to the server
            base URL
        query:
        json: JSON data to be conveyed in the body of the request
        data: Form data to be conveyed
            in the body of the request
        multipart: If `True`, `data` should be conveyed
            as a multipart request.
        query: A dictionary from which to assemble the
            query string.
        headers:
        multipart_data_headers:
        timeout:
    """
    # For backwards compatibility...
    if isinstance(data, (str, bytes, sob.abc.Model)) or (data is None):
        json = data
        data = ()
    request: typing.Callable[
        [
            str,
            str,
            str | bytes | sob.abc.Model | None,
            collections.abc.Mapping[str, sob.abc.MarshallableTypes]
            | collections.abc.Sequence[
                tuple[str, sob.abc.MarshallableTypes]
            ],
            collections.abc.Mapping[str, sob.abc.MarshallableTypes]
            | collections.abc.Sequence[
                tuple[str, sob.abc.MarshallableTypes]
            ]
            | str,
            collections.abc.Mapping[str, sob.abc.MarshallableTypes]
            | collections.abc.Sequence[
                tuple[str, sob.abc.MarshallableTypes]
            ],
            bool,
            collections.abc.Mapping[
                str, collections.abc.MutableMapping[str, str]
            ]
            | collections.abc.Sequence[
                tuple[str, collections.abc.MutableMapping[str, str]]
            ],
            int,
        ],
        sob.abc.Readable,
    ] = self._request
    # Wrap the _request method with a retry decorator if more
    # than one attempt is specified in the config
    if self.retry_number_of_attempts > 1:
        request = retry(
            errors=self.retry_for_errors,
            number_of_attempts=self.retry_number_of_attempts,
            retry_hook=self.retry_hook,
            logger=self.logger,
        )(self._request)
    return request(
        path,
        method,
        json,
        data,
        query,
        headers,
        multipart,
        multipart_data_headers,
        timeout,
    )

ClientModule

ClientModule(
    open_api: (
        str | sob.abc.Readable | oapi.oas.model.OpenAPI
    ),
    model_path: str | pathlib.Path,
    *,
    class_name: str = "Client",
    base_class: type[
        oapi.client.Client
    ] = oapi.client.Client,
    imports: str | tuple[str, ...] = (),
    init_decorator: str = "",
    include_init_parameters: str | tuple[str, ...] = (),
    add_init_parameters: str | tuple[str, ...] = (),
    add_init_parameter_docs: str | tuple[str, ...] = (),
    init_parameter_defaults: (
        collections.abc.Mapping[str, typing.Any]
        | collections.abc.Sequence[tuple[str, typing.Any]]
    ) = (),
    init_parameter_defaults_source: (
        collections.abc.Mapping[str, typing.Any]
        | collections.abc.Sequence[tuple[str, typing.Any]]
    ) = (),
    get_method_name_from_path_method_operation: typing.Callable[
        [str, str, str | None], str
    ] = oapi.client.get_default_method_name_from_path_method_operation,
    use_operation_id: bool = False
)

This class parses an Open API document and outputs a module defining a client class for interfacing with the API described by an OpenAPI document.

Parameters:

• open_api (str | sob.abc.Readable | oapi.oas.model.OpenAPI) –

  An OpenAPI document. This can be a URL, file-path, an HTTP response (http.client.HTTPResponse), a file object, or an instance of oapi.oas.model.OpenAPI.

• model_path (str | pathlib.Path) –

  The file path where the data model for this client can be found. This must be a model generated using oapi.model, and must be part of the same project that this client will be saved into.

• class_name (str, default: 'Client' ) –
• base_class (type[oapi.client.Client], default: oapi.client.Client ) –

  The base class to use for the client. If provided, this must be a sub-class of oapi.client.Client.

• imports (str | tuple[str, ...], default: () ) –

  One or more import statements to include (in addition to those generated automatically).

• init_decorator (str, default: '' ) –

  A decorator to apply to the client class .__init__ method. If used, make sure to include any modules referenced in your imports. For example: "@decorator_function(argument_a=1, argument_b=2)".

• include_init_parameters (str | tuple[str, ...], default: () ) –

  The name of all parameters to include for the client's .__init__ method.

• add_init_parameters (str | tuple[str, ...], default: () ) –

  Additional parameter declarations to add to the client's .__init__ method. These should include a type hint and default value (not just the parameter name). For example: 'additional_parameter_name: str | None = None'. Note: additional parameters will not do anything without the use of a decorator which utilizes the additional parameters, so use of this parameter should be accompanied by an init_decorator.

• add_init_parameter_docs (str | tuple[str, ...], default: () ) –
• init_parameter_defaults (collections.abc.Mapping[str, typing.Any] | collections.abc.Sequence[tuple[str, typing.Any]], default: () ) –

  A mapping of parameter names to default values for the parameter.

• init_parameter_defaults_source (collections.abc.Mapping[str, typing.Any] | collections.abc.Sequence[tuple[str, typing.Any]], default: () ) –

  A mapping of parameter names to default values for the parameter expressed as source code. This is to allow for the passing of imported constants, expressions, etc.

Source code in src/oapi/client.py

def __init__(
    self,
    open_api: str | sob.abc.Readable | OpenAPI,
    model_path: str | Path,
    *,
    class_name: str = "Client",
    base_class: type[Client] = Client,
    imports: str | tuple[str, ...] = (),
    init_decorator: str = "",
    include_init_parameters: str | tuple[str, ...] = (),
    add_init_parameters: str | tuple[str, ...] = (),
    add_init_parameter_docs: str | tuple[str, ...] = (),
    init_parameter_defaults: collections.abc.Mapping[str, typing.Any]
    | collections.abc.Sequence[tuple[str, typing.Any]] = (),
    init_parameter_defaults_source: collections.abc.Mapping[
        str, typing.Any
    ]
    | collections.abc.Sequence[tuple[str, typing.Any]] = (),
    get_method_name_from_path_method_operation: typing.Callable[
        [str, str, str | None], str
    ] = get_default_method_name_from_path_method_operation,
    use_operation_id: bool = False,
) -> None:
    """
    Parameters:
        open_api: An OpenAPI document. This can be a URL, file-path, an
            HTTP response (`http.client.HTTPResponse`), a file object, or
            an instance of `oapi.oas.model.OpenAPI`.
        model_path: The file path where the data model for this
            client can be found. This must be a model generated using
            `oapi.model`, and must be part of the same project that this
            client will be saved into.
        class_name:
        base_class: The base class to use for the client. If provided,
            this must be a sub-class of `oapi.client.Client`.
        imports: One or more import statements to include
            (in addition to those generated automatically).
        init_decorator:  A decorator to apply to the client class
            `.__init__` method. If used, make sure to include any modules
            referenced in your `imports`. For example:
            "@decorator_function(argument_a=1, argument_b=2)".
        include_init_parameters: The name of all parameters to
            include for the client's `.__init__` method.
        add_init_parameters: Additional parameter
            declarations to add to the client's `.__init__` method.
            These should include a type hint and default value (not just
            the parameter name). For example:
            'additional_parameter_name: str | None = None'. Note:
            additional parameters will not do anything without the use of a
            decorator which utilizes the additional parameters, so use of
            this parameter should be accompanied by an `init_decorator`.
        add_init_parameter_docs:
        init_parameter_defaults: A mapping of
            parameter names to default values for the parameter.
        init_parameter_defaults_source: A mapping of
            parameter names to default values for the parameter *expressed
            as source code*. This is to allow for the passing of imported
            constants, expressions, etc.
    """
    message: str
    if isinstance(model_path, Path):
        model_path = str(model_path)
    # Ensure a valid model path has been provided
    if not os.path.exists(model_path):
        raise ValueError(model_path)
    # Get an OpenAPI document
    if isinstance(open_api, str):
        if os.path.exists(open_api):
            open_api = _get_path_open_api(open_api)
        else:
            open_api = _get_url_open_api(open_api)
    elif isinstance(open_api, sob.abc.Readable):
        open_api = _get_io_open_api(open_api)
    elif not isinstance(open_api, OpenAPI):
        message = (
            f"`{sob.utilities.get_calling_function_qualified_name()}` "
            "requires an instance of `str`, "
            f"`{sob.utilities.get_qualified_name(OpenAPI)}`, or a "
            "file-like object for the `open_api` parameter--not "
            f"{open_api!r}"
        )
        raise TypeError(message)
    # Ensure all elements have URLs and JSON pointers
    sob.set_model_url(open_api, sob.get_model_url(open_api))
    sob.set_model_pointer(open_api, sob.get_model_pointer(open_api) or "")
    self.open_api: OpenAPI = open_api
    init_decorator = init_decorator.strip()
    if init_decorator and not init_decorator.startswith("@"):
        raise ValueError(init_decorator)
    self._init_decorator: str = init_decorator.strip()
    self._include_init_parameters: tuple[str, ...] = (
        (include_init_parameters,)
        if isinstance(include_init_parameters, str)
        else include_init_parameters
    )
    self._add_init_parameters: tuple[str, ...] = (
        (add_init_parameters,)
        if isinstance(add_init_parameters, str)
        else add_init_parameters
    )
    self._add_init_parameter_docs: tuple[str, ...] = (
        (add_init_parameter_docs,)
        if isinstance(add_init_parameter_docs, str)
        else add_init_parameter_docs
    )
    self._init_parameter_defaults: dict[str, typing.Any] = dict(
        init_parameter_defaults
    )
    self._init_parameter_defaults_source: dict[str, typing.Any] = dict(
        init_parameter_defaults_source
    )
    self._resolver: Resolver = Resolver(open_api)
    self._model_path: str = model_path
    self._base_class: type[Client] = base_class
    self._class_name: str = class_name
    # This keeps track of used names in the global namespace
    self._names: set[str] = {
        "sob",
        "oapi",
        "typing",
        "Logger",
        self._class_name,
    }
    self._imports: set[str] = {
        "from __future__ import annotations",
        "import typing",
        "import oapi",
        "import sob",
        "from logging import Logger",
    } | set(
        filter(None, (imports,) if isinstance(imports, str) else imports)
    )
    if "headers" not in self._iter_excluded_parameter_names():
        # `collections.abc` is only referenced if `headers` is used
        self._imports.add("import collections.abc")
        self._names.add("collections")
    self.get_method_name_from_path_method_operation = (
        get_method_name_from_path_method_operation
    )
    self.use_operation_id = use_operation_id

get_source

get_source(path: str | pathlib.Path) -> str

This generates source code for the client module.

Parameters:

• path (str | pathlib.Path) –

  The file path for the client module. This is needed in order to determine the relative import for the model module.

Source code in src/oapi/client.py

def get_source(self, path: str | Path) -> str:
    """
    This generates source code for the client module.

    Parameters:
        path: The file path for the client module. This is needed
            in order to determine the relative import for the model module.
    """
    source: str = "\n".join(
        chain(
            self._iter_class_declaration_source(),
            self._iter_init_method_source(),
            self._iter_paths_operations_methods_source(),
            ("",),
        )
    )
    # The model module import couldn't be generated until we
    # knew the client module path, so we add it now
    self._imports |= {
        self._get_model_module_import(path),
        self._get_client_base_class_import(path),
    }
    # Imports need to be generated last
    imports: str = "\n".join(self._iter_imports())
    return sob.utilities.suffix_long_lines(
        f"{imports}\n\n\n{source.rstrip()}\n"
    )

save

save(path: str | pathlib.Path) -> None

This method will save the generated module to a given path.

Source code in src/oapi/client.py

def save(self, path: str | Path) -> None:
    """
    This method will save the generated module to a given path.
    """
    client_source: str = self.get_source(path=path)
    # Save the module
    with open(path, "w") as model_io:
        model_io.write(client_source)

urlencode

urlencode(
    query: (
        collections.abc.Mapping[str, typing.Any]
        | sob.abc.Dictionary
        | sob.abc.Object
        | collections.abc.Sequence[tuple[str, typing.Any]]
    ),
    doseq: bool = True,
    safe: str = oapi.client.URLENCODE_SAFE,
    encoding: str = "utf-8",
    errors: str = "",
    quote_via: typing.Callable[
        [str, bytes | str, str, str], str
    ] = urllib.parse.quote,
) -> str

This function wraps urllib.parse.urlencode, but has different default argument values. Additionally, when a mapping/dictionary is passed for a query value, that dictionary/mapping performs an update to the query dictionary.

Parameters:

• query (collections.abc.Mapping[str, typing.Any] | sob.abc.Dictionary | sob.abc.Object | collections.abc.Sequence[tuple[str, typing.Any]]) –
• doseq (bool, default: True ) –
• safe (str, default: oapi.client.URLENCODE_SAFE ) –
• encoding (str, default: 'utf-8' ) –
• errors (str, default: '' ) –
• quote_via (typing.Callable[[str, bytes | str, str, str], str], default: urllib.parse.quote ) –

Source code in src/oapi/client.py

def urlencode(
    query: collections.abc.Mapping[str, typing.Any]
    | sob.abc.Dictionary
    | sob.abc.Object
    | collections.abc.Sequence[tuple[str, typing.Any]],
    doseq: bool = True,  # noqa: FBT001 FBT002
    safe: str = URLENCODE_SAFE,
    encoding: str = "utf-8",
    errors: str = "",
    quote_via: typing.Callable[[str, bytes | str, str, str], str] = quote,
) -> str:
    """
    This function wraps `urllib.parse.urlencode`, but has different default
    argument values. Additionally, when a mapping/dictionary is passed for a
    query value, that dictionary/mapping performs an update to the query
    dictionary.

    Parameters:
        query:
        doseq:
        safe:
        encoding:
        errors:
        quote_via:
    """
    items: list[tuple[str, typing.Any]] = []
    key: str
    value: typing.Any
    for key, value in _iter_items(query):
        if isinstance(value, _ITEMIZED_TYPES):
            # The only dictionaries/objects which should exist
            # for values which have been formatted are
            # intended to be bumped up into the top-level
            items.extend(_iter_items(value))
        else:
            items.append((key, value))
    return _urlencode(
        query=items,
        doseq=doseq,
        safe=safe,
        encoding=encoding,
        errors=errors,
        quote_via=quote_via,
    )

format_argument_value

format_argument_value(
    name: str,
    value: sob.abc.MarshallableTypes | typing.IO[bytes],
    style: str,
    *,
    explode: bool = False,
    multipart: bool = False
) -> (
    str
    | dict[str, str]
    | dict[str, str | list[str]]
    | collections.abc.Sequence[str]
    | collections.abc.Sequence[bytes]
    | bytes
    | collections.abc.Sequence[typing.IO[bytes]]
    | typing.IO[bytes]
    | None
)

Format an argument value for use in a path, query, cookie, header, etc.

Parameters:

• value (sob.abc.MarshallableTypes | typing.IO[bytes]) –
• style (str) –
• explode (bool, default: False ) –
• multipart (bool, default: False ) –

  Indicates the argument will be part of a multipart request

See: https://swagger.io/docs/specification/serialization/

Source code in src/oapi/client.py

def format_argument_value(  # noqa: C901
    name: str,
    value: sob.abc.MarshallableTypes | typing.IO[bytes],
    style: str,
    *,
    explode: bool = False,
    multipart: bool = False,
) -> (
    str
    | dict[str, str]
    | dict[str, str | list[str]]
    | collections.abc.Sequence[str]
    | collections.abc.Sequence[bytes]
    | bytes
    | collections.abc.Sequence[typing.IO[bytes]]
    | typing.IO[bytes]
    | None
):
    """
    Format an argument value for use in a path, query, cookie, header, etc.

    Parameters:
        value:
        style:
        explode:
        multipart: Indicates the argument will be part of a
            multipart request

    See: https://swagger.io/docs/specification/serialization/
    """
    if multipart and (
        isinstance(value, (bytes, sob.abc.Readable))
        or (
            value
            and (not isinstance(value, str))
            and isinstance(value, collections.abc.Sequence)
            and isinstance(value[0], (bytes, sob.abc.Readable))
        )
    ):
        # For multipart requests, we don't apply any formatting to `bytes`
        # objects, as they will be sent in binary format
        return value
    if isinstance(value, sob.abc.Model):
        value = sob.marshal(value)  # type: ignore
    if style == "simple":
        return _format_simple_argument_value(value, explode=explode)
    if style == "label":
        return _format_label_argument_value(value, explode=explode)
    if style == "matrix":
        return _format_matrix_argument_value(name, value, explode=explode)
    if style == "form":
        return _format_form_argument_value(value, explode=explode)
    if style == "spaceDelimited":
        return _format_space_delimited_argument_value(value, explode=explode)
    if style == "pipeDelimited":
        return _format_pipe_delimited_argument_value(value, explode=explode)
    if style == "deepObject":
        return _format_deep_object_argument_value(name, value, explode=explode)
    if style == "dotObject":
        return _format_dot_object_argument_value(name, value, explode=explode)
    raise ValueError(style)

get_request_curl

get_request_curl(
    request: urllib.request.Request,
    options: str = "-i",
    censored_headers: collections.abc.Iterable[str] = (
        "X-api-key",
        "Authorization",
    ),
    censored_parameters: tuple[str, ...] = (
        "client_secret",
    ),
) -> str

Render an instance of urllib.request.Request as a curl command.

Parameters:

• options (str, default: '-i' ) –

  Any additional parameters to pass to curl, (such as "--compressed", "--insecure", etc.)

Source code in src/oapi/client.py

def get_request_curl(
    request: Request,
    options: str = "-i",
    censored_headers: collections.abc.Iterable[str] = (
        "X-api-key",
        "Authorization",
    ),
    censored_parameters: tuple[str, ...] = ("client_secret",),
) -> str:
    """
    Render an instance of `urllib.request.Request` as a `curl` command.

    Parameters:
        options: Any additional parameters to pass to `curl`,
            (such as "--compressed", "--insecure", etc.)
    """
    is_json: bool = bool(
        request.headers.get("Content-Type", "").lower() == "application/json"
    )
    censored_headers = tuple(map(str.lower, censored_headers))

    def _get_request_curl_header_item(item: tuple[str, str]) -> str:
        key: str
        value: str
        key, value = item
        if key.lower() in censored_headers:
            value = "***"
        return "-H {}".format(shlex.quote(f"{key}: {value}"))

    data: bytes = (
        request.data
        if isinstance(request.data, bytes)
        else (
            request.data.read()  # type: ignore
            if isinstance(request.data, sob.abc.Readable)
            else (
                b"".join(request.data) if request.data else b""  # type: ignore
            )
        )
    )
    repr_data: str = ""
    if data:
        data_str = data.decode("utf-8")
        if censored_parameters and not is_json:
            query: dict[str, list[str]] = {}
            key: str
            value: list[str]
            for key, value in parse_qs(data_str).items():
                if key in censored_parameters:
                    query[key] = ["***"]
                else:
                    query[key] = value
            data_str = urlencode(query, safe=f"*{URLENCODE_SAFE}", doseq=True)
        try:
            repr_data = shlex.quote(data_str)
        except UnicodeDecodeError:
            # If the data can't be represented as a command-line argument,
            # use a placeholder
            repr_data = "***"
    return " ".join(
        filter(
            None,
            (
                f"curl -X {request.get_method()}",
                options,
                " ".join(
                    map(
                        _get_request_curl_header_item,
                        sorted(request.headers.items()),  # type: ignore
                    )
                ),
                (f"-d {repr_data}" if data else ""),
                shlex.quote(request.full_url),
            ),
        )
    )

default_retry_hook

default_retry_hook(error: Exception) -> bool

By default, don't retry for HTTP 404 (NOT FOUND) errors and HTTP 401 (UNAUTHORIZED) errors.

Source code in src/oapi/client.py

def default_retry_hook(error: Exception) -> bool:
    """
    By default, don't retry for HTTP 404 (NOT FOUND) errors
    and HTTP 401 (UNAUTHORIZED) errors.
    """
    if isinstance(error, HTTPError) and error.code in (404, 401, 409, 410):
        return False
    return True

retry

retry(
    errors: (
        tuple[type[Exception], ...] | type[Exception]
    ) = Exception,
    retry_hook: typing.Callable[
        [Exception], bool
    ] = oapi.client.default_retry_hook,
    number_of_attempts: int = 1,
    logger: logging.Logger | None = None,
) -> typing.Callable

This function decorates another, and causes the decorated function to be re-attempted a specified number of times, with exponential backoff, until the decorated function is successful or the maximum number of attempts is reached (in which case an exception is raised).

Parameters:

• errors (tuple[type[Exception], ...] | type[Exception], default: Exception ) –

  A sub-class of Exception, or a tuple of one or more sub-classes of Exception. The default is Exception, causing all errors to trigger a retry.

• retry_hook (typing.Callable[[Exception], bool], default: oapi.client.default_retry_hook ) –

  A function accepting as its only argument the handled exception, and returning a boolean value indicating whether or not to retry the function.

• number_of_attempts (int, default: 1 ) –

  The maximum number of times to attempt execution of the function, including the first execution. Please note that, because the default for this parameter is 1, this decorator will do nothing if this argument is not provided.

Source code in src/oapi/client.py

def retry(
    errors: tuple[type[Exception], ...] | type[Exception] = Exception,
    retry_hook: typing.Callable[[Exception], bool] = default_retry_hook,
    number_of_attempts: int = 1,
    logger: Logger | None = None,
) -> typing.Callable:
    """
    This function decorates another, and causes the decorated function
    to be re-attempted a specified number of times, with exponential
    backoff, until the decorated function is successful or the maximum
    number of attempts is reached (in which case an exception is raised).

    Parameters:
        errors: A sub-class of `Exception`, or a tuple of one or more
            sub-classes of `Exception`. The default is `Exception`, causing
            *all* errors to trigger a retry.
        retry_hook: A function accepting as its only argument the handled
            exception, and returning a boolean value indicating whether or not
            to retry the function.
        number_of_attempts: The maximum number of times to attempt
            execution of the function, *including* the first execution. Please
            note that, because the default for this parameter is 1, this
            decorator will do *nothing* if this argument is not provided.
    """

    def decorating_function(function: typing.Callable) -> typing.Callable:
        attempt_number: int = 1

        @functools.wraps(function)
        def wrapper(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
            nonlocal attempt_number
            nonlocal number_of_attempts
            if number_of_attempts - attempt_number:
                try:
                    return function(*args, **kwargs)
                except errors as error:
                    if not retry_hook(error):
                        raise
                    warning_message: str = (
                        f"Attempt # {attempt_number!s}:\n"
                        f"{sob.errors.get_exception_text()}"
                    )
                    warn(warning_message, stacklevel=2)
                    if logger is not None:
                        logger.warning(warning_message)
                    sleep(2**attempt_number)
                    attempt_number += 1
                    return wrapper(*args, **kwargs)
            return function(*args, **kwargs)

        return wrapper

    return decorating_function

write_client_module

write_client_module(
    client_path: str | pathlib.Path,
    *,
    open_api: (
        str | sob.abc.Readable | oapi.oas.model.OpenAPI
    ),
    model_path: str | pathlib.Path,
    class_name: str = "Client",
    base_class: type[
        oapi.client.Client
    ] = oapi.client.Client,
    imports: str | tuple[str, ...] = (),
    init_decorator: str = "",
    include_init_parameters: str | tuple[str, ...] = (),
    add_init_parameters: str | tuple[str, ...] = (),
    add_init_parameter_docs: str | tuple[str, ...] = (),
    init_parameter_defaults: (
        collections.abc.Mapping[str, typing.Any]
        | collections.abc.Sequence[tuple[str, typing.Any]]
    ) = (),
    init_parameter_defaults_source: (
        collections.abc.Mapping[str, typing.Any]
        | collections.abc.Sequence[tuple[str, typing.Any]]
    ) = (),
    get_method_name_from_path_method_operation: typing.Callable[
        [str, str, str | None], str
    ] = oapi.client.get_default_method_name_from_path_method_operation,
    use_operation_id: bool = False
) -> None

This function parses an Open API document and outputs a module defining a client class for interfacing with the API described by an OpenAPI document.

Parameters:

• client_path (str | pathlib.Path) –

  The file path where the client module will be saved (created or updated).

• open_api (str | sob.abc.Readable | oapi.oas.model.OpenAPI) –

  An OpenAPI document. This can be a URL, file-path, an HTTP response (http.client.HTTPResponse), a file object, or an instance of oapi.oas.model.OpenAPI.

• model_path (str | pathlib.Path) –

  The file path where the data model for this client can be found. This must be a model generated using oapi.model, and must be part of the same project that this client will be saved into.

• class_name (str, default: 'Client' ) –
• base_class (type[oapi.client.Client], default: oapi.client.Client ) –

  The base class to use for the client. If provided, this must be a sub-class of oapi.client.Client.

• imports (str | tuple[str, ...], default: () ) –

  One or more import statements to include (in addition to those generated automatically).

• init_decorator (str, default: '' ) –

  A decorator to apply to the client class .__init__ method. If used, make sure to include any modules referenced in your imports. For example: "@decorator_function(argument_a=1, argument_b=2)".

• include_init_parameters (str | tuple[str, ...], default: () ) –

  The name of all parameters to include for the client's .__init__ method.

• add_init_parameters (str | tuple[str, ...], default: () ) –

  Additional parameter declarations to add to the client's .__init__ method. These should include a type hint and default value (not just the parameter name). For example: 'additional_parameter_name: str | None = None'. Note: additional parameters will not do anything without the use of a decorator which utilizes the additional parameters, so use of this parameter should be accompanied by an init_decorator.

• add_init_parameter_docs (str | tuple[str, ...], default: () ) –
• init_parameter_defaults (collections.abc.Mapping[str, typing.Any] | collections.abc.Sequence[tuple[str, typing.Any]], default: () ) –

  A mapping of parameter names to default values for the parameter.

• init_parameter_defaults_source (collections.abc.Mapping[str, typing.Any] | collections.abc.Sequence[tuple[str, typing.Any]], default: () ) –

  A mapping of parameter names to default values for the parameter expressed as source code. This is to allow for the passing of imported constants, expressions, etc.

Source code in src/oapi/client.py

def write_client_module(
    client_path: str | Path,
    *,
    open_api: str | sob.abc.Readable | OpenAPI,
    model_path: str | Path,
    class_name: str = "Client",
    base_class: type[Client] = Client,
    imports: str | tuple[str, ...] = (),
    init_decorator: str = "",
    include_init_parameters: str | tuple[str, ...] = (),
    add_init_parameters: str | tuple[str, ...] = (),
    add_init_parameter_docs: str | tuple[str, ...] = (),
    init_parameter_defaults: collections.abc.Mapping[str, typing.Any]
    | collections.abc.Sequence[tuple[str, typing.Any]] = (),
    init_parameter_defaults_source: collections.abc.Mapping[str, typing.Any]
    | collections.abc.Sequence[tuple[str, typing.Any]] = (),
    get_method_name_from_path_method_operation: typing.Callable[
        [str, str, str | None], str
    ] = get_default_method_name_from_path_method_operation,
    use_operation_id: bool = False,
) -> None:
    """
    This function parses an Open API document and outputs a module defining
    a client class for interfacing with the API described by an OpenAPI
    document.

    Parameters:
        client_path: The file path where the client module will be saved
            (created or updated).
        open_api: An OpenAPI document. This can be a URL, file-path, an
            HTTP response (`http.client.HTTPResponse`), a file object, or
            an instance of `oapi.oas.model.OpenAPI`.
        model_path: The file path where the data model for this
            client can be found. This must be a model generated using
            `oapi.model`, and must be part of the same project that this
            client will be saved into.
        class_name:
        base_class: The base class to use for the client. If provided,
            this must be a sub-class of `oapi.client.Client`.
        imports: One or more import statements to include
            (in addition to those generated automatically).
        init_decorator:  A decorator to apply to the client class
            `.__init__` method. If used, make sure to include any modules
            referenced in your `imports`. For example:
            "@decorator_function(argument_a=1, argument_b=2)".
        include_init_parameters: The name of all parameters to
            include for the client's `.__init__` method.
        add_init_parameters: Additional parameter
            declarations to add to the client's `.__init__` method.
            These should include a type hint and default value (not just
            the parameter name). For example:
            'additional_parameter_name: str | None = None'. Note:
            additional parameters will not do anything without the use of a
            decorator which utilizes the additional parameters, so use of
            this parameter should be accompanied by an `init_decorator`.
        add_init_parameter_docs:
        init_parameter_defaults: A mapping of
            parameter names to default values for the parameter.
        init_parameter_defaults_source: A mapping of
            parameter names to default values for the parameter *expressed
            as source code*. This is to allow for the passing of imported
            constants, expressions, etc.
    """
    locals_: dict[str, typing.Any] = dict(locals())
    locals_.pop("client_path")
    client_module: ClientModule = ClientModule(**locals_)
    client_module.save(client_path)