mirror of
https://gitlab.com/wgp/dougal/software.git
synced 2025-12-06 10:27:09 +00:00
177 lines
5.4 KiB
Python
177 lines
5.4 KiB
Python
import os
|
|
import pathlib
|
|
from glob import glob
|
|
from yaml import full_load as _load
|
|
|
|
"""
|
|
Interface to the instance configuration.
|
|
|
|
The instance configuration is expected to be found under
|
|
$HOME/etc/config.yaml and the configuration for the individual
|
|
surveys should be under $HOME/etc/surveys/*.yaml. In both cases,
|
|
$HOME is the home directory of the user running this script.
|
|
"""
|
|
|
|
def is_relative_to(it, other):
|
|
"""
|
|
is_relative_to() is not present version Python 3.9, so we
|
|
need this kludge to get Dougal to run on OpenSUSE 15.4
|
|
"""
|
|
|
|
if "is_relative_to" in dir(it):
|
|
return it.is_relative_to(other)
|
|
|
|
return str(it.absolute()).startswith(str(other.absolute()))
|
|
|
|
|
|
prefix = os.environ.get("DOUGAL_ROOT", os.environ.get("HOME", ".")+"/software")
|
|
|
|
DOUGAL_ROOT = os.environ.get("DOUGAL_ROOT", os.environ.get("HOME", ".")+"/software")
|
|
VARDIR = os.environ.get("VARDIR", DOUGAL_ROOT+"/var")
|
|
LOCKFILE = os.environ.get("LOCKFILE", VARDIR+"/runner.lock")
|
|
|
|
def vars ():
|
|
return {
|
|
"DOUGAL_ROOT": DOUGAL_ROOT,
|
|
"VARDIR": VARDIR,
|
|
"LOCKFILE": LOCKFILE
|
|
}
|
|
|
|
def read (file = None):
|
|
if file is None:
|
|
file = prefix+"/etc/config.yaml"
|
|
with open(file, "r") as fd:
|
|
cfg = _load(fd)
|
|
|
|
if "db" in cfg and "connection_string" in cfg["db"] and os.environ.get("PGDATABASE"):
|
|
# This should cause the PostgreSQL library to use the environment variables
|
|
cfg["db"]["connection_string"] = ""
|
|
print("Using environment variables for database connection")
|
|
|
|
return cfg
|
|
|
|
def files (globspec = None, include_archived = False):
|
|
"""
|
|
Read and parse survey configuration files.
|
|
|
|
Arguments:
|
|
|
|
globspec (string): a glob spec matching the selection of files to read.
|
|
If not provided, a default value will be used.
|
|
|
|
include_archived (bool): whether to include surveys marked as archived or
|
|
not. Defaults to ignoring archived surveys.
|
|
|
|
Note that file names starting with `.` or `_` (dot or underscore) will
|
|
never be read (and there is no option to override this). Intended for
|
|
quickly and temporarily “disabling” a survey configuration by renaming
|
|
the relevant file.
|
|
"""
|
|
|
|
print("This method is obsolete")
|
|
return
|
|
|
|
tuples = []
|
|
|
|
if globspec is None:
|
|
globspec = prefix+'/etc/surveys/*.yaml'
|
|
|
|
for filepath in glob(globspec):
|
|
filepath = os.path.abspath(filepath)
|
|
if os.path.basename(filepath).startswith("_") or os.path.basename(filepath).startswith("."):
|
|
continue
|
|
|
|
survey = read(filepath)
|
|
|
|
if not include_archived and "archived" in survey and survey["archived"] is not False:
|
|
continue
|
|
|
|
tuples.append((filepath, survey))
|
|
|
|
return tuples
|
|
|
|
def surveys (globspec = None, include_archived = False):
|
|
return [i[1] for i in files(globspec, include_archived)]
|
|
|
|
def rxflags (flagstr):
|
|
"""
|
|
Convert flags string into a Python flags argument.
|
|
"""
|
|
flags = 0
|
|
cases = {
|
|
"i": re.I
|
|
}
|
|
for flag in flagstr:
|
|
flags |= cases.get(flag, 0)
|
|
return flags
|
|
|
|
def translate_path (file):
|
|
"""
|
|
Translate a path from a Dougal import directory to an actual
|
|
physical path on disk.
|
|
|
|
Any user files accessible by Dougal must be under a path prefixed
|
|
by `(config.yaml).imports.paths`. The value of `imports.paths` may
|
|
be either a string, in which case this represents the prefix under
|
|
which all Dougal data resides, or a dictionary where the keys are
|
|
logical paths and their values the corresponding physical path.
|
|
"""
|
|
cfg = read()
|
|
root = pathlib.Path(DOUGAL_ROOT)
|
|
filepath = pathlib.Path(file).resolve()
|
|
import_paths = cfg["imports"]["paths"]
|
|
|
|
if filepath.is_absolute():
|
|
if type(import_paths) == str:
|
|
# Substitute the root for the real physical path
|
|
# NOTE: `root` deals with import_paths not being absolute
|
|
prefix = root.joinpath(pathlib.Path(import_paths)).resolve()
|
|
return str(pathlib.Path(prefix).joinpath(*filepath.parts[2:]))
|
|
else:
|
|
# Look for a match on the second path element
|
|
if filepath.parts[1] in import_paths:
|
|
# NOTE: `root` deals with import_paths[…] not being absolute
|
|
prefix = root.joinpath(import_paths[filepath.parts[1]])
|
|
return str(pathlib.Path(prefix).joinpath(*filepath.parts[2:]))
|
|
else:
|
|
# This path is invalid
|
|
raise TypeError("invalid path or file: {0!r}".format(filepath))
|
|
else:
|
|
# A relative filepath is always resolved relative to the logical root
|
|
root = pathlib.Path("/")
|
|
return translate_path(root.joinpath(filepath))
|
|
|
|
def untranslate_path (file):
|
|
"""
|
|
Attempt to convert a physical path into a logical one.
|
|
See `translate_path()` above for details.
|
|
"""
|
|
cfg = read()
|
|
dougal_root = pathlib.Path(DOUGAL_ROOT)
|
|
filepath = pathlib.Path(file).resolve()
|
|
import_paths = cfg["imports"]["paths"]
|
|
physical_root = pathlib.Path("/")
|
|
|
|
if filepath.is_absolute():
|
|
if type(import_paths) == str:
|
|
if is_relative_to(filepath, import_paths):
|
|
physical_root = pathlib.Path("/")
|
|
physical_prefix = pathlib.Path(import_paths)
|
|
return str(root.joinpath(filepath.relative_to(physical_prefix)))
|
|
else:
|
|
raise TypeError("invalid path or file: {0!r}".format(filepath))
|
|
else:
|
|
for key, value in import_paths.items():
|
|
value = dougal_root.joinpath(value)
|
|
physical_prefix = pathlib.Path(value)
|
|
if is_relative_to(filepath, physical_prefix):
|
|
logical_prefix = physical_root.joinpath(pathlib.Path(key)).resolve()
|
|
return str(logical_prefix.joinpath(filepath.relative_to(physical_prefix)))
|
|
|
|
# If we got here with no matches, this is not a valid
|
|
# Dougal data path
|
|
raise TypeError("invalid path or file: {0!r}".format(filepath))
|
|
else:
|
|
# A relative filepath is always resolved relative to DOUGAL_ROOT
|
|
return untranslate_path(root.joinpath(filepath))
|