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.")