Rework `.config` routines to use `pathlib.Path`
Been meaning to do this port for a while and since it makes passing around file handles (presumably alongside the in mem obj form) a lot simpler/nicer and the implementations of all the config file handling much more terse with less presumptions about the form of filename/dir `str` values all over the place B) moar technically, let's us: - drop remaining `.config` usage of `os.path`. - return `Path`s from most routines. - adds a special case to `get_conf_path()` such that if the input name contains a `pps.` pattern, we avoid validating the name; this is going to be used by new `.accounting.open_pps()` code which will instead write a separate TOML file for each account B)rekt_pps
parent
bc249fbeca
commit
7b3d724908
114
piker/config.py
114
piker/config.py
|
@ -21,8 +21,6 @@ Platform configuration (files) mgmt.
|
||||||
import platform
|
import platform
|
||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
from os import path
|
|
||||||
from os.path import dirname
|
|
||||||
import shutil
|
import shutil
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
@ -126,30 +124,35 @@ def get_app_dir(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
_config_dir = _click_config_dir = get_app_dir('piker')
|
_click_config_dir: Path = Path(get_app_dir('piker'))
|
||||||
_parent_user = os.environ.get('SUDO_USER')
|
_config_dir: Path = _click_config_dir
|
||||||
|
_parent_user: str = os.environ.get('SUDO_USER')
|
||||||
|
|
||||||
if _parent_user:
|
if _parent_user:
|
||||||
non_root_user_dir = os.path.expanduser(
|
non_root_user_dir = Path(
|
||||||
f'~{_parent_user}'
|
os.path.expanduser(f'~{_parent_user}')
|
||||||
)
|
)
|
||||||
root = 'root'
|
root: str = 'root'
|
||||||
|
_ccds: str = str(_click_config_dir) # click config dir string
|
||||||
|
i_tail: int = int(_ccds.rfind(root) + len(root))
|
||||||
_config_dir = (
|
_config_dir = (
|
||||||
non_root_user_dir +
|
non_root_user_dir
|
||||||
_click_config_dir[
|
/
|
||||||
_click_config_dir.rfind(root) + len(root):
|
Path(_ccds[i_tail+1:]) # +1 to capture trailing '/'
|
||||||
]
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
_conf_names: set[str] = {
|
_conf_names: set[str] = {
|
||||||
'brokers',
|
'brokers',
|
||||||
'pps',
|
# 'pps',
|
||||||
'trades',
|
'trades',
|
||||||
'watchlists',
|
'watchlists',
|
||||||
'paper_trades'
|
'paper_trades'
|
||||||
}
|
}
|
||||||
|
|
||||||
_watchlists_data_path = os.path.join(_config_dir, 'watchlists.json')
|
# TODO: probably drop all this super legacy, questrade specific,
|
||||||
|
# config stuff XD ?
|
||||||
|
_watchlists_data_path: Path = _config_dir / Path('watchlists.json')
|
||||||
_context_defaults = dict(
|
_context_defaults = dict(
|
||||||
default_map={
|
default_map={
|
||||||
# Questrade specific quote poll rates
|
# Questrade specific quote poll rates
|
||||||
|
@ -180,7 +183,7 @@ def _conf_fn_w_ext(
|
||||||
def get_conf_path(
|
def get_conf_path(
|
||||||
conf_name: str = 'brokers',
|
conf_name: str = 'brokers',
|
||||||
|
|
||||||
) -> str:
|
) -> Path:
|
||||||
'''
|
'''
|
||||||
Return the top-level default config path normally under
|
Return the top-level default config path normally under
|
||||||
``~/.config/piker`` on linux for a given ``conf_name``, the config
|
``~/.config/piker`` on linux for a given ``conf_name``, the config
|
||||||
|
@ -196,72 +199,68 @@ def get_conf_path(
|
||||||
- strats.toml
|
- strats.toml
|
||||||
|
|
||||||
'''
|
'''
|
||||||
assert conf_name in _conf_names
|
if 'pps.' not in conf_name:
|
||||||
|
assert str(conf_name) in _conf_names
|
||||||
|
|
||||||
fn = _conf_fn_w_ext(conf_name)
|
fn = _conf_fn_w_ext(conf_name)
|
||||||
return os.path.join(
|
return _config_dir / Path(fn)
|
||||||
_config_dir,
|
|
||||||
fn,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def repodir():
|
def repodir() -> Path:
|
||||||
'''
|
'''
|
||||||
Return the abspath to the repo directory.
|
Return the abspath as ``Path`` to the git repo's root dir.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
dirpath = path.abspath(
|
return Path(__file__).absolute().parent.parent
|
||||||
# we're 3 levels down in **this** module file
|
|
||||||
dirname(dirname(os.path.realpath(__file__)))
|
|
||||||
)
|
|
||||||
return dirpath
|
|
||||||
|
|
||||||
|
|
||||||
def load(
|
def load(
|
||||||
conf_name: str = 'brokers',
|
conf_name: str = 'brokers',
|
||||||
path: str = None,
|
path: Path | None = None,
|
||||||
|
|
||||||
**tomlkws,
|
**tomlkws,
|
||||||
|
|
||||||
) -> (dict, str):
|
) -> tuple[dict, str]:
|
||||||
'''
|
'''
|
||||||
Load config file by name.
|
Load config file by name.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
path = path or get_conf_path(conf_name)
|
path: Path = path or get_conf_path(conf_name)
|
||||||
|
|
||||||
if not os.path.isdir(_config_dir):
|
if not _config_dir.is_dir():
|
||||||
Path(_config_dir).mkdir(parents=True, exist_ok=True)
|
_config_dir.mkdir(
|
||||||
|
parents=True,
|
||||||
if not os.path.isfile(path):
|
exist_ok=True,
|
||||||
fn = _conf_fn_w_ext(conf_name)
|
|
||||||
|
|
||||||
template = os.path.join(
|
|
||||||
repodir(),
|
|
||||||
'config',
|
|
||||||
fn
|
|
||||||
)
|
)
|
||||||
# try to copy in a template config to the user's directory
|
|
||||||
# if one exists.
|
if not path.is_file():
|
||||||
if os.path.isfile(template):
|
fn: str = _conf_fn_w_ext(conf_name)
|
||||||
|
|
||||||
|
# try to copy in a template config to the user's directory if
|
||||||
|
# one exists.
|
||||||
|
template: Path = repodir() / 'config' / fn
|
||||||
|
if template.is_file():
|
||||||
shutil.copyfile(template, path)
|
shutil.copyfile(template, path)
|
||||||
else:
|
else:
|
||||||
# create an empty file
|
# create empty file
|
||||||
with open(path, 'x'):
|
with path.open(mode='x'):
|
||||||
pass
|
pass
|
||||||
else:
|
else:
|
||||||
with open(path, 'r'):
|
with path.open(mode='r'):
|
||||||
pass # touch it
|
pass # touch it
|
||||||
|
|
||||||
config = toml.load(path, **tomlkws)
|
config: dict = toml.load(str(path), **tomlkws)
|
||||||
log.debug(f"Read config file {path}")
|
log.debug(f"Read config file {path}")
|
||||||
return config, path
|
return config, path
|
||||||
|
|
||||||
|
|
||||||
def write(
|
def write(
|
||||||
config: dict, # toml config as dict
|
config: dict, # toml config as dict
|
||||||
name: str = 'brokers',
|
|
||||||
path: str = None,
|
name: str | None = None,
|
||||||
|
path: Path | None = None,
|
||||||
fail_empty: bool = True,
|
fail_empty: bool = True,
|
||||||
|
|
||||||
**toml_kwargs,
|
**toml_kwargs,
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
|
@ -271,21 +270,26 @@ def write(
|
||||||
Create a ``brokers.ini`` file if one does not exist.
|
Create a ``brokers.ini`` file if one does not exist.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
path = path or get_conf_path(name)
|
if name:
|
||||||
dirname = os.path.dirname(path)
|
path: Path = path or get_conf_path(name)
|
||||||
if not os.path.isdir(dirname):
|
dirname: Path = path.parent
|
||||||
|
if not dirname.is_dir():
|
||||||
log.debug(f"Creating config dir {_config_dir}")
|
log.debug(f"Creating config dir {_config_dir}")
|
||||||
os.makedirs(dirname)
|
dirname.mkdir()
|
||||||
|
|
||||||
if not config and fail_empty:
|
if (
|
||||||
|
not config
|
||||||
|
and fail_empty
|
||||||
|
):
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Watch out you're trying to write a blank config!")
|
"Watch out you're trying to write a blank config!"
|
||||||
|
)
|
||||||
|
|
||||||
log.debug(
|
log.debug(
|
||||||
f"Writing config `{name}` file to:\n"
|
f"Writing config `{name}` file to:\n"
|
||||||
f"{path}"
|
f"{path}"
|
||||||
)
|
)
|
||||||
with open(path, 'w') as cf:
|
with path.open(mode='w') as cf:
|
||||||
return toml.dump(
|
return toml.dump(
|
||||||
config,
|
config,
|
||||||
cf,
|
cf,
|
||||||
|
|
Loading…
Reference in New Issue