auth_cds

d4science_copernicus/auth_cds.py

@@ -0,0 +1,203 @@
+import os
+import cdsapi
+
+CONFIG_FILE = ".cdsapirc"
+
+CURRENT_CDSAPIRC = "./" + CONFIG_FILE
+HOME_CDSAPIRC = "~/" + CONFIG_FILE
+
+DEFAULT_URL = 'https://cds-beta.climate.copernicus.eu/api'
+
+ENV_CDSAPI_URL = 'CDSAPI_URL'
+ENV_CDSAPI_KEY = 'CDSAPI_KEY'
+
+CONFIG_URL = 'url'
+CONFIG_KEY = 'key'
+
+
+def create_config(url, key):
+    return {'url': url, 'key': key}
+
+
+def config_from_env():
+    config = {}
+    url = os.environ.get(ENV_CDSAPI_URL)
+    key = os.environ.get(ENV_CDSAPI_KEY)
+
+    if url is not None:
+        config[CONFIG_URL] = url
+
+    if key is not None:
+        config[CONFIG_KEY] = key
+
+    if url is not None and key is not None:
+        print("config from environment %s - %s" %
+                  (ENV_CDSAPI_URL, ENV_CDSAPI_KEY))
+        return config
+
+    return None
+
+
+def config_to_env(config):
+    os.environ[ENV_CDSAPI_URL] = config[CONFIG_URL]
+    os.environ[ENV_CDSAPI_KEY] = config[CONFIG_KEY]
+    print("set envs %s , %s" % (ENV_CDSAPI_URL, ENV_CDSAPI_KEY))
+
+
+def config_to_file(config, config_path=HOME_CDSAPIRC):
+    URL = config[CONFIG_URL]
+    KEY = config[CONFIG_KEY]
+    with open(config_path, 'w') as file:
+        file.write(f"url: {URL}\n")
+        file.write(f"key: {KEY}\n")
+    print("saved config file %s" % config_path)
+
+
+def config_from_file(custom_path=None):
+    config = None
+
+    check_paths = [CURRENT_CDSAPIRC, HOME_CDSAPIRC]
+
+    # Path of the .cdsapirc file
+    if custom_path:
+        check_paths = [custom_path]
+
+    for path in check_paths:
+        # Check if the file exists
+        if os.path.exists(path):
+            config = cdsapi.api.read_config(path)
+            print("config from file %s" % path)
+            break
+
+    if not config:
+        return None
+
+    return config
+
+
+def authenticate(set_environment=True):
+    # config from OS
+    config = config_from_env()
+
+    if not config:
+        # config from file
+        config = config_from_file()
+
+    if not config:
+        config = query_credentials()
+
+    if set_environment and config is not None:
+        config_to_env(config)
+
+    return config
+
+
+def input_url():
+    URL = input(f"Insert URL. Leave empty for default {
+                DEFAULT_URL}: ") or DEFAULT_URL
+    return URL
+
+
+def input_key():
+    KEY = input("Insert your key: ")
+    return KEY
+
+
+def query_url():
+    URL = input_url()
+    os.environ[ENV_CDSAPI_URL] = URL
+
+
+def query_key():
+    KEY = input_key()
+    os.environ[ENV_CDSAPI_KEY] = KEY
+
+
+def query_save(config):
+    save_to_file = input(
+        "Do you want to save the URL and KEY to the .cdsapirc file? (h)ome / (c)urrent / path. empty to skip ")
+    save_to_file = save_to_file.strip()
+
+    if save_to_file == "":
+        print("configuration not saved.")
+        return None
+
+    if save_to_file.lower().strip() == "h":
+        save_to_file = HOME_CDSAPIRC
+
+    elif save_to_file.lower().strip() == "c":
+        save_to_file = CURRENT_CDSAPIRC
+
+    else:
+        save_to_file + "/" + CONFIG_FILE
+
+    config_to_file(config, save_to_file)
+    print("URL and KEY saved to %s" % save_to_file)
+
+    return config
+
+
+def query_credentials():
+    URL = input_url()
+    KEY = input_key()
+    config = {'url': URL, 'key': KEY}
+    query_save(config)
+    return config
+
+
+def show_credentials(verbose=True):
+    # config from ENV
+    config_env = None
+    config_file = None
+
+    config_env = config_from_env()
+
+    if verbose:
+        print("config from env: %s" % config_env)
+    
+    # if not verbose, we just need to find the first one
+    # if verbose, we need to print all the existing configs
+    elif config_env is not None:
+        return config_env
+
+    config_file = config_from_file()
+    if verbose:
+         print("config from file: %s" % config_file)
+    elif config_file is not None:
+        return config_file
+
+    config = config_env or config_file
+    if verbose:
+        if config is None:
+            print("cannot obtain configuration")
+        else:
+            print("current config: %s" % config)
+
+    return config
+
+
+def removeCredentials(from_home=True, from_current=True, custom_path=None):
+    if from_home and os.path.exists(HOME_CDSAPIRC):
+        print("removing configuration %s" % HOME_CDSAPIRC)
+        os.remove(HOME_CDSAPIRC)
+
+    if from_current and os.path.exists(CURRENT_CDSAPIRC):
+        print("removing configuration %s" % CURRENT_CDSAPIRC)
+        os.remove(CURRENT_CDSAPIRC)
+
+    if custom_path is not None and os.path.exists(custom_path):
+        print("removing configuration %s" % custom_path)
+        os.remove(custom_path)
+
+
+def getAuthenticatedClient():
+    config = authenticate()
+    if config is None:
+        print("cannot obtain authentication")
+        return None
+
+    c = cdsapi.Client(url=config.get("url"), key=config.get("key"))
+    return c
+
+# config from file ./.cdsapirc
+# {'url': 'https://cds-beta.climate.copernicus.eu/api', 'key': 'db1f2085-6b8b-42e6-b832-625dfaf831a4'}