| from __future__ import annotations |
|
|
| import configparser |
| import json |
| import os |
| import warnings |
| from typing import Any |
|
|
| conf: dict[str, dict[str, Any]] = {} |
| default_conf_dir = os.path.join(os.path.expanduser("~"), ".config/fsspec") |
| conf_dir = os.environ.get("FSSPEC_CONFIG_DIR", default_conf_dir) |
|
|
|
|
| def set_conf_env(conf_dict, envdict=os.environ): |
| """Set config values from environment variables |
| |
| Looks for variables of the form ``FSSPEC_<protocol>`` and |
| ``FSSPEC_<protocol>_<kwarg>``. For ``FSSPEC_<protocol>`` the value is parsed |
| as a json dictionary and used to ``update`` the config of the |
| corresponding protocol. For ``FSSPEC_<protocol>_<kwarg>`` there is no |
| attempt to convert the string value, but the kwarg keys will be lower-cased. |
| |
| The ``FSSPEC_<protocol>_<kwarg>`` variables are applied after the |
| ``FSSPEC_<protocol>`` ones. |
| |
| Parameters |
| ---------- |
| conf_dict : dict(str, dict) |
| This dict will be mutated |
| envdict : dict-like(str, str) |
| Source for the values - usually the real environment |
| """ |
| kwarg_keys = [] |
| for key in envdict: |
| if key.startswith("FSSPEC_") and len(key) > 7 and key[7] != "_": |
| if key.count("_") > 1: |
| kwarg_keys.append(key) |
| continue |
| try: |
| value = json.loads(envdict[key]) |
| except json.decoder.JSONDecodeError as ex: |
| warnings.warn( |
| f"Ignoring environment variable {key} due to a parse failure: {ex}" |
| ) |
| else: |
| if isinstance(value, dict): |
| _, proto = key.split("_", 1) |
| conf_dict.setdefault(proto.lower(), {}).update(value) |
| else: |
| warnings.warn( |
| f"Ignoring environment variable {key} due to not being a dict:" |
| f" {type(value)}" |
| ) |
| elif key.startswith("FSSPEC"): |
| warnings.warn( |
| f"Ignoring environment variable {key} due to having an unexpected name" |
| ) |
|
|
| for key in kwarg_keys: |
| _, proto, kwarg = key.split("_", 2) |
| conf_dict.setdefault(proto.lower(), {})[kwarg.lower()] = envdict[key] |
|
|
|
|
| def set_conf_files(cdir, conf_dict): |
| """Set config values from files |
| |
| Scans for INI and JSON files in the given dictionary, and uses their |
| contents to set the config. In case of repeated values, later values |
| win. |
| |
| In the case of INI files, all values are strings, and these will not |
| be converted. |
| |
| Parameters |
| ---------- |
| cdir : str |
| Directory to search |
| conf_dict : dict(str, dict) |
| This dict will be mutated |
| """ |
| if not os.path.isdir(cdir): |
| return |
| allfiles = sorted(os.listdir(cdir)) |
| for fn in allfiles: |
| if fn.endswith(".ini"): |
| ini = configparser.ConfigParser() |
| ini.read(os.path.join(cdir, fn)) |
| for key in ini: |
| if key == "DEFAULT": |
| continue |
| conf_dict.setdefault(key, {}).update(dict(ini[key])) |
| if fn.endswith(".json"): |
| with open(os.path.join(cdir, fn)) as f: |
| js = json.load(f) |
| for key in js: |
| conf_dict.setdefault(key, {}).update(dict(js[key])) |
|
|
|
|
| def apply_config(cls, kwargs, conf_dict=None): |
| """Supply default values for kwargs when instantiating class |
| |
| Augments the passed kwargs, by finding entries in the config dict |
| which match the classes ``.protocol`` attribute (one or more str) |
| |
| Parameters |
| ---------- |
| cls : file system implementation |
| kwargs : dict |
| conf_dict : dict of dict |
| Typically this is the global configuration |
| |
| Returns |
| ------- |
| dict : the modified set of kwargs |
| """ |
| if conf_dict is None: |
| conf_dict = conf |
| protos = cls.protocol if isinstance(cls.protocol, (tuple, list)) else [cls.protocol] |
| kw = {} |
| for proto in protos: |
| |
| if proto in conf_dict: |
| kw.update(conf_dict[proto]) |
| |
| kw.update(**kwargs) |
| kwargs = kw |
| return kwargs |
|
|
|
|
| set_conf_files(conf_dir, conf) |
| set_conf_env(conf) |
|
|
