Module ciocore.config¶

A configuration object implemented as a module-level singleton.

Configuration variables can be shared by importing the module. If there are changes in environment variables or other sources, the config can be refreshed. See config()

Functions¶

config¶

Source

def config(force=False):


    global __config__
    if force or not __config__:
        __config__ = Config()
    return __config__

config(force=False)

DEPRECATED! Use get() instead.

Instantiate a config object if necessary.

Keyword Arguments:

force -- Discards any existing config object and instantiate a new one -- Defaults to False.

Returns:

Config() object containing a dictionary of configuration variables.

get¶

Source

def get(force=False):

    global __config__
    if force or not __config__:
        __config__ = Config()
    return __config__.config

get(force=False)

Instantiate a config object if necessary and return the dictionary.

Keyword Arguments:

force -- Discards any existing config object and instantiate a new one -- Defaults to False.

Returns:

A dictionary containing configuration.

Example

from ciocore import config

print(config.get())

# Result:
# {
#   'thread_count': 16,
#   'priority': 5,
#   'md5_caching': True,
#   'log_level': 'INFO',
#   'url': 'https://dashboard.conductortech.com',
#   'auth_url': 'https://dashboard.conductortech.com',
#   'api_url': 'https://api.conductortech.com', 
#   'api_key': None
# }

Classes¶

Config¶

Source

def __init__(self):


    try:
        default_thread_count = min(multiprocessing.cpu_count() * 2, 16)
    except NotImplementedError:
        default_thread_count = 16

    url = os.environ.get("CONDUCTOR_URL", "https://dashboard.conductortech.com")

    if not URL_REGEX.match(url):
        raise ValueError("CONDUCTOR_URL is not valid '{}'".format(url))

    api_url = os.environ.get("CONDUCTOR_API_URL", url.replace("dashboard", "api"))
    if not URL_REGEX.match(api_url):
        raise ValueError("CONDUCTOR_API_URL is not valid '{}'".format(api_url))

    falsy = ["false", "no", "off", "0"]

    log_level = os.environ.get("CONDUCTOR_LOG_LEVEL", "INFO")
    if log_level not in ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"]:
        log_level = "INFO"

    self.config = {
        "thread_count": int(os.environ.get("CONDUCTOR_THREAD_COUNT", default_thread_count)),
        "priority": int(os.environ.get("CONDUCTOR_PRIORITY", 5)),
        "md5_caching": False
        if os.environ.get("CONDUCTOR_MD5_CACHING", "True").lower() in falsy
        else True,
        "log_level": log_level,
        "url": url,
        # Keep "auth_url" for backwwards compatibillity only. 
        # Clients should use "url" moving forward. 
        # Remove "auth_url" on the next major version bump.
        "auth_url": url,
        "api_url": api_url,
        "api_key": self.get_api_key_from_variable() or self.get_api_key_from_file(),
    }

Config()

Create a configuration dictionary.

Raises:

ValueError -- Invalid inputs, such as badly formed URLs.

Static methods¶

get_api_key_from_variable¶

Source

@staticmethod
def get_api_key_from_variable():

    api_key = os.environ.get("CONDUCTOR_API_KEY")
    if not api_key:
        return
    logger.info("Attempting to read API key from CONDUCTOR_API_KEY")
    try:
        return json.loads(api_key.replace("\n", "").replace("\r", ""))
    except ValueError:
        try:
            result = base64.b64decode(api_key)
            return Config._to_json(result)
        except BaseException:
            result = base64.b64decode(api_key.encode()).decode("ascii")
            return Config._to_json(result)
    except BaseException:
        message = "An error occurred reading the API key from the CONDUCTOR_API_KEY variable"
        logger.error(message)
        raise ValueError(message)

get_api_key_from_variable()

Attempt to get an API key from the CONDUCTOR_API_KEY environment variable.

Raises:

ValueError -- An error occurred while loading the key into JSON.

Returns:

JSON object containing the key - base 64 decoded if necessary.

get_api_key_from_file¶

Source

@staticmethod
def get_api_key_from_file():

    api_key_path = os.environ.get("CONDUCTOR_API_KEY_PATH")
    if not api_key_path:
        return
    logger.info("Attempting to read API key from CONDUCTOR_API_KEY_PATH")
    try:
        with open(api_key_path, "r") as fp:
            return Config._to_json(fp.read())
    except BaseException:
        message = "An error occurred reading the API key from the path described in the CONDUCTOR_API_KEY_PATH variable"
        logger.error(message)
        raise ValueError(message)

get_api_key_from_file()

Attempt to get an API key from the file in the CONDUCTOR_API_KEY_PATH environment variable.

Raises:

ValueError -- An error occurred while reading or loading the key into JSON.

Returns:

JSON object containing the key - base 64 decoded if necessary.