utils.py

import inspect
import logging
import httpx
from taskingai.client.api_client import SyncApiClient, AsyncApiClient
from enum import Enum


def check_kwargs(caller, given):
    argspec = inspect.getfullargspec(caller)
    diff = set(given).difference(argspec.args)
    if diff:
        logging.exception(caller.__name__ + " had unexpected keyword argument(s): " + ", ".join(diff), exc_info=False)


def get_version():
    from taskingai import __version__

    return __version__


def get_user_agent():
    client_id = f"python-client-{get_version()}"
    user_agent_details = {"httpx": httpx.__version__}
    user_agent = "{} ({})".format(client_id, ", ".join([f"{k}:{v}" for k, v in user_agent_details.items()]))
    return user_agent


def convert_query_params_dict(query_params_dict):
    """
    This helper function will convert the query_params_dict to a format where enum values are converted to their
    corresponding string values.
    """
    converted_dict = {}
    for key, value in query_params_dict.items():
        if isinstance(value, Enum):
            converted_dict[key] = value.value
        elif isinstance(value, bool):
            converted_dict[key] = str(value).lower()
        elif value is not None:
            converted_dict[key] = value
    return converted_dict


__sync_api_client = None
__async_api_client = None


def get_api_client(async_client: bool):
    global __sync_api_client, __async_api_client

    from taskingai.config import Config

    client_config = Config.OPENAPI_CONFIG
    client_config.api_key = client_config.api_key or {}

    if async_client:
        if not __async_api_client:
            __async_api_client = AsyncApiClient(configuration=client_config)
            __async_api_client.user_agent = get_user_agent()
        return __async_api_client

    else:
        if not __sync_api_client:
            __sync_api_client = SyncApiClient(configuration=client_config)
            __sync_api_client.user_agent = get_user_agent()
        return __sync_api_client