def load_config()

in classify-split-extract-workflow/classify-job/config.py [0:0]

• 32 lines of code
• 7 McCabe index (conditional complexity)

def load_config(bucket_name: str, filename: str) -> Optional[Dict[Any, Any]]:
    """
    Loads the configuration data from a GCS bucket or local file.

    Args:
        bucket_name (str): The GCS bucket name.
        filename (str): The configuration file name.

    Returns:
        Optional[Dict[Any, Any]]: The configuration data.
    """
    global BUCKET, CONFIG_DATA, LAST_MODIFIED_TIME_OF_CONFIG
    if not BUCKET:
        BUCKET = init_bucket(bucket_name)

    if not BUCKET:
        return None

    blob = BUCKET.get_blob(filename)
    if not blob:
        logger.error(f"Error: file does not exist gs://{bucket_name}/{filename}")
        return None

    last_modified_time = blob.updated
    if LAST_MODIFIED_TIME_OF_CONFIG == last_modified_time:
        return CONFIG_DATA

    logger.info(f"Reloading config from: {filename}")
    try:
        CONFIG_DATA = json.loads(blob.download_as_text(encoding="utf-8"))
        LAST_MODIFIED_TIME_OF_CONFIG = last_modified_time
    except (json.JSONDecodeError, OSError) as e:
        logger.error(
            f"Error while obtaining file from GCS gs://{bucket_name}/{filename}: {e}"
        )
        logger.warning(f"Using local {filename}")
        try:
            with open(
                os.path.join(os.path.dirname(__file__), "config", filename),
                encoding="utf-8",
            ) as json_file:
                CONFIG_DATA = json.load(json_file)
        except (FileNotFoundError, json.JSONDecodeError) as exc:
            logger.error(f"Error loading local config file {filename}: {exc}")
            return None

    return CONFIG_DATA