Source code for openghg.util._user

import logging
import os
import platform
from pathlib import Path
from typing import Dict, Optional
import uuid
import toml
import shutil
from openghg.types import ConfigFileError

logger = logging.getLogger("openghg.util")
logger.setLevel(logging.DEBUG)  # Have to set level for logger as well as handler

openghg_config_filename = "openghg.conf"


# @lru_cache
[docs] def get_user_id() -> str: """Return the user's ID Returns: str: User ID """ config = read_local_config() uid = str(config.get("user_id", "NA")) return uid
def get_default_objectstore_path() -> Path: """Returns the default object store path in the user's home directory Returns: Path: Object store path in ~/openghg_store """ return Path.home().joinpath("openghg_store").absolute() # @lru_cache
[docs] def get_user_config_path() -> Path: """Returns path to user config file. This file is created in the user's home directory in ~/.ghgconfig/openghg/user.conf on Linux / macOS or in LOCALAPPDATA/openghg/openghg.conf on Windows. Returns: pathlib.Path: Path to user config file """ user_platform = platform.system() if user_platform == "Windows": appdata_path = os.getenv("LOCALAPPDATA") if appdata_path is None: raise ValueError("Unable to read LOCALAPPDATA environment variable.") config_path = Path(appdata_path).joinpath("openghg", openghg_config_filename) elif user_platform in ("Linux", "Darwin"): config_path = Path.home().joinpath(".openghg", openghg_config_filename) else: raise ValueError(f"Unknown platform: {user_platform}") return config_path
[docs] def create_config(silent: bool = False) -> None: """Creates a user config. Args: silent: Creates the basic configuration file with only the user's object store in a default location. Returns: None """ if not silent: print("\nOpenGHG configuration") print("---------------------\n") user_config_path = get_user_config_path() # Current config version as of version 0.6.0 config_version = "2" positive_responses = ("y", "yes") # If the config file exists we might need to update it due to the introduction # of the user ID and new object store path handling for multiple stores if user_config_path.exists(): if silent: logger.error("Cannot update an existing configuration silently. Please run interactively.") return logger.info(f"User config exists at {str(user_config_path)}, checking...") config = toml.loads(user_config_path.read_text()) recent = "config_version" in config if recent: user_store_path = Path(config["object_store"]["user"]["path"]) else: user_store_path = Path(config["object_store"]["local_store"]) logger.info(f"Current user object store path: {user_store_path}") # Store the object store info stores = {} update_input = input("Would you like to update the path? (y/n): ") if update_input.lower() in positive_responses: new_path_input = input("Enter new path for object store: ") if not new_path_input: print("You must enter a path. Unable to complete config setup.") return new_path = Path(new_path_input).expanduser().resolve() stores["user"] = {"path": str(new_path), "permissions": "rw"} else: stores["user"] = {"path": str(user_store_path), "permissions": "rw"} # Copy in exisiting shared stores if recent: stores.update({k: v for k, v in config["object_store"].items() if k != "user"}) # Now ask the user if they want to add new stores new_shared_stores = _user_multstore_input() if new_shared_stores: existing = [k for k in new_shared_stores if k in stores] if existing: print(f"Some names match those of existing stores: {existing}, please update manually.") stores.update(new_shared_stores) # Some users may not have a user ID if they've used previous versions of OpenGHG user_id = config.get("user_id") config = _combine_config(config_version=config_version, object_stores=stores, user_id=user_id) else: # Let's try migrating the old config # If it works we call this function again # otherwise continue on to create a new config try: _migrate_config() except FileNotFoundError: pass else: create_config(silent=silent) return stores = {} # 1. Create the user's object store first if not silent: logger.info("We'll first create your user object store.\n") obj_store_path = get_default_objectstore_path() if silent: obj_store_path = get_default_objectstore_path() else: obj_store_input = input(f"Enter path for your local object store (default {obj_store_path}): ") if obj_store_input: obj_store_path = Path(obj_store_input).expanduser().resolve() # Let's create the store to make sure it's a valid path obj_store_path.mkdir(parents=True, exist_ok=True) stores["user"] = {"path": str(obj_store_path), "permissions": "rw"} if not silent: shared_stores = _user_multstore_input() stores.update(shared_stores) config = _combine_config(config_version=config_version, object_stores=stores) # Make the .config/openghg folder user_config_path.parent.mkdir(parents=True, exist_ok=True) if not silent: logger.info(f"Configuration written to {user_config_path}") user_config_path.write_text(toml.dumps(config))
def _user_multstore_input() -> Dict: """Ask the user to input data about shared object stores Returns: dict: Dictionary of object store paths and permissions """ positive_responses = ("y", "yes") stores = {} # 2. Ask the user to enter other object store paths while True: response = input("Would you like to add another object store? (y/n): ") if response.lower() in positive_responses: store_name = input("Enter the name of the store: ") store_path = input("Enter the object store path: ") print("\nYou will now be asked for read/write permissions for the store.") print("For read only enter r, for read and write enter rw.") store_permissions = "" while store_permissions not in ("r", "rw"): store_permissions = input("\nEnter object store permissions: ") stores[store_name] = {"path": store_path, "permissions": store_permissions} else: break return stores def _combine_config(config_version: str, object_stores: Dict, user_id: Optional[str] = None) -> Dict: """Combine parts required into the proper dictionary format Args: config_version: Configuration version number object_stores: Object store configuration dictionary user_id: User ID """ # Create the object store dictionary object_store_info = {} for name, data in object_stores.items(): path = str(Path(data["path"]).expanduser().resolve()) permissions = data["permissions"].strip() object_store_info[name] = {"path": path, "permissions": permissions} if user_id is None: user_id = str(uuid.uuid4()) return {"user_id": user_id, "config_version": config_version, "object_store": object_store_info} # @lru_cache
[docs] def read_local_config() -> Dict: """Reads the local config file. Returns: dict: OpenGHG configurations """ config_path = get_user_config_path() if not config_path.exists(): try: _migrate_config() except FileNotFoundError as e: raise ConfigFileError( "Unable to read configuration file, please see the installation instructions \ or run openghg --quickstart" ) from e config: Dict = toml.loads(config_path.read_text()) try: _ = config["object_store"]["user"] except KeyError: raise ConfigFileError( "Invalid config file detected, please please see the installation instructions \ or run openghg --quickstart" ) return config
[docs] def check_config() -> None: """Check that the user config file is valid and the paths given in it exist. Raises ConfigFileError if problems found. Returns: None """ config_path = get_user_config_path() please_update = "please run openghg --quickstart to update it." if not config_path.exists(): raise ConfigFileError( "Configuration file does not exist. Please create it by running openghg --quickstart." ) config = read_local_config() try: uid = config["user_id"] except KeyError: raise ConfigFileError("Unable to read user ID, ") try: uuid.UUID(uid, version=4) except ValueError: raise ConfigFileError(f"Invalid user ID, {please_update}") try: _ = config["config_version"] except KeyError: raise ConfigFileError(f"Invalid config file, {please_update}") try: object_stores = config["object_store"] except KeyError: raise ConfigFileError(f"Unable to read object store data, {please_update}") for name, data in object_stores.items(): p = Path(data["path"]) if not p.exists(): logger.info(f"The path for object store {name} at {p} does not exist but will be created.")
def _migrate_config() -> None: """If user config file is in ~/.config, move it to ~/.openghg. If no config is found in ~/.config or system is Windows, raise FileNotFoundError. Returns: None """ old_config_path = Path.home().joinpath(".config", "openghg", openghg_config_filename) if old_config_path.exists(): new_config_path = get_user_config_path() new_config_path.parent.mkdir(parents=True) shutil.move(str(old_config_path), str(new_config_path)) logger.info(f"Moved user config file from {str(old_config_path)} to {str(new_config_path)}.") shutil.rmtree(old_config_path.parent) # remove "openghg" dir from ~/.config else: raise FileNotFoundError("Configuration file not found.")