elasticsearch_serverless/_async/client/_base.py [32:119]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
)
from elastic_transport.client_utils import DEFAULT, DefaultType

from ..._otel import OpenTelemetry
from ...compat import warn_stacklevel
from ...exceptions import (
    HTTP_EXCEPTIONS,
    ApiError,
    ElasticsearchWarning,
    UnsupportedProductError,
)
from .utils import _base64_auth_header, _quote_query

_WARNING_RE = re.compile(r"\"([^\"]*)\"")


def resolve_auth_headers(
    headers: Optional[Mapping[str, str]],
    http_auth: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    api_key: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    basic_auth: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    bearer_auth: Union[DefaultType, None, str] = DEFAULT,
) -> HttpHeaders:
    if headers is None:
        headers = HttpHeaders()
    elif not isinstance(headers, HttpHeaders):
        headers = HttpHeaders(headers)

    resolved_http_auth = http_auth if http_auth is not DEFAULT else None
    resolved_basic_auth = basic_auth if basic_auth is not DEFAULT else None
    if resolved_http_auth is not None:
        if resolved_basic_auth is not None:
            raise ValueError(
                "Can't specify both 'http_auth' and 'basic_auth', "
                "instead only specify 'basic_auth'"
            )
        if isinstance(http_auth, str) or (
            isinstance(resolved_http_auth, (list, tuple))
            and all(isinstance(x, str) for x in resolved_http_auth)
        ):
            resolved_basic_auth = resolved_http_auth
        else:
            raise TypeError(
                "The deprecated 'http_auth' parameter must be either 'Tuple[str, str]' or 'str'. "
                "Use either the 'basic_auth' parameter instead"
            )

        warnings.warn(
            "The 'http_auth' parameter is deprecated. "
            "Use 'basic_auth' or 'bearer_auth' parameters instead",
            category=DeprecationWarning,
            stacklevel=warn_stacklevel(),
        )

    resolved_api_key = api_key if api_key is not DEFAULT else None
    resolved_bearer_auth = bearer_auth if bearer_auth is not DEFAULT else None
    if resolved_api_key or resolved_basic_auth or resolved_bearer_auth:
        if (
            sum(
                x is not None
                for x in (
                    resolved_api_key,
                    resolved_basic_auth,
                    resolved_bearer_auth,
                )
            )
            > 1
        ):
            raise ValueError(
                "Can only set one of 'api_key', 'basic_auth', and 'bearer_auth'"
            )
        if headers and headers.get("authorization", None) is not None:
            raise ValueError(
                "Can't set 'Authorization' HTTP header with other authentication options"
            )
        if resolved_api_key:
            headers["authorization"] = f"ApiKey {_base64_auth_header(resolved_api_key)}"
        if resolved_basic_auth:
            headers["authorization"] = (
                f"Basic {_base64_auth_header(resolved_basic_auth)}"
            )
        if resolved_bearer_auth:
            headers["authorization"] = f"Bearer {resolved_bearer_auth}"

    return headers


class BaseClient:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



elasticsearch_serverless/_sync/client/_base.py [32:119]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
)
from elastic_transport.client_utils import DEFAULT, DefaultType

from ..._otel import OpenTelemetry
from ...compat import warn_stacklevel
from ...exceptions import (
    HTTP_EXCEPTIONS,
    ApiError,
    ElasticsearchWarning,
    UnsupportedProductError,
)
from .utils import _base64_auth_header, _quote_query

_WARNING_RE = re.compile(r"\"([^\"]*)\"")


def resolve_auth_headers(
    headers: Optional[Mapping[str, str]],
    http_auth: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    api_key: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    basic_auth: Union[DefaultType, None, Tuple[str, str], str] = DEFAULT,
    bearer_auth: Union[DefaultType, None, str] = DEFAULT,
) -> HttpHeaders:
    if headers is None:
        headers = HttpHeaders()
    elif not isinstance(headers, HttpHeaders):
        headers = HttpHeaders(headers)

    resolved_http_auth = http_auth if http_auth is not DEFAULT else None
    resolved_basic_auth = basic_auth if basic_auth is not DEFAULT else None
    if resolved_http_auth is not None:
        if resolved_basic_auth is not None:
            raise ValueError(
                "Can't specify both 'http_auth' and 'basic_auth', "
                "instead only specify 'basic_auth'"
            )
        if isinstance(http_auth, str) or (
            isinstance(resolved_http_auth, (list, tuple))
            and all(isinstance(x, str) for x in resolved_http_auth)
        ):
            resolved_basic_auth = resolved_http_auth
        else:
            raise TypeError(
                "The deprecated 'http_auth' parameter must be either 'Tuple[str, str]' or 'str'. "
                "Use either the 'basic_auth' parameter instead"
            )

        warnings.warn(
            "The 'http_auth' parameter is deprecated. "
            "Use 'basic_auth' or 'bearer_auth' parameters instead",
            category=DeprecationWarning,
            stacklevel=warn_stacklevel(),
        )

    resolved_api_key = api_key if api_key is not DEFAULT else None
    resolved_bearer_auth = bearer_auth if bearer_auth is not DEFAULT else None
    if resolved_api_key or resolved_basic_auth or resolved_bearer_auth:
        if (
            sum(
                x is not None
                for x in (
                    resolved_api_key,
                    resolved_basic_auth,
                    resolved_bearer_auth,
                )
            )
            > 1
        ):
            raise ValueError(
                "Can only set one of 'api_key', 'basic_auth', and 'bearer_auth'"
            )
        if headers and headers.get("authorization", None) is not None:
            raise ValueError(
                "Can't set 'Authorization' HTTP header with other authentication options"
            )
        if resolved_api_key:
            headers["authorization"] = f"ApiKey {_base64_auth_header(resolved_api_key)}"
        if resolved_basic_auth:
            headers["authorization"] = (
                f"Basic {_base64_auth_header(resolved_basic_auth)}"
            )
        if resolved_bearer_auth:
            headers["authorization"] = f"Bearer {resolved_bearer_auth}"

    return headers


class BaseClient:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -