Remove old config code

This commit is contained in:
Tim Van Baak 2021-06-13 18:04:46 -07:00
parent bea38d67df
commit 7e3db84b3c
5 changed files with 0 additions and 413 deletions

View File

@ -1,23 +0,0 @@
# Module imports
from .dict import AttrOrderedDict, ReadOnlyOrderedDict
from .directory import (
RootConfigDirectoryContext,
UserConfigDirectoryContext,
LexiconConfigDirectoryContext,
is_guid)
# Environment variable name constants
ENV_SECRET_KEY = "AMANUENSIS_SECRET_KEY"
ENV_CONFIG_DIR = "AMANUENSIS_CONFIG_DIR"
ENV_LOG_FILE = "AMANUENSIS_LOG_FILE"
ENV_LOG_FILE_SIZE = "AMANUENSIS_LOG_FILE_SIZE"
ENV_LOG_FILE_NUM = "AMANUENSIS_LOG_FILE_NUM"
__all__ = [
AttrOrderedDict.__name__,
ReadOnlyOrderedDict.__name__,
RootConfigDirectoryContext.__name__,
UserConfigDirectoryContext.__name__,
LexiconConfigDirectoryContext.__name__,
is_guid.__name__,
]

View File

@ -1,82 +0,0 @@
"""
`with` context managers for mediating config file access.
"""
# Standard library imports
import fcntl
import json
# Application imports
from .dict import AttrOrderedDict, ReadOnlyOrderedDict
class open_lock():
"""A context manager that opens a file with the specified file lock"""
def __init__(self, path, mode, lock_type):
self.fd = open(path, mode, encoding='utf8')
fcntl.lockf(self.fd, lock_type)
def __enter__(self):
return self.fd
def __exit__(self, exc_type, exc_value, traceback):
fcntl.lockf(self.fd, fcntl.LOCK_UN)
self.fd.close()
class open_sh(open_lock):
"""A context manager that opens a file with a shared lock"""
def __init__(self, path, mode):
super().__init__(path, mode, fcntl.LOCK_SH)
class open_ex(open_lock):
"""A context manager that opens a file with an exclusive lock"""
def __init__(self, path, mode):
super().__init__(path, mode, fcntl.LOCK_EX)
class json_ro(open_sh):
"""
A context manager that opens a file in a shared, read-only mode.
The contents of the file are read as JSON and returned as a read-
only OrderedDict.
"""
def __init__(self, path):
super().__init__(path, 'r')
self.config = None
def __enter__(self) -> ReadOnlyOrderedDict:
self.config = json.load(self.fd, object_pairs_hook=ReadOnlyOrderedDict)
return self.config
class json_rw(open_ex):
"""
A context manager that opens a file with an exclusive lock. The
file mode defaults to r+, which requires that the file exist. The
file mode can be set to w+ to create a new file by setting the new
kwarg in the ctor. The contents of the file are read as JSON and
returned in an AttrOrderedDict. Any changes to the context dict
will be written out to the file when the context manager exits,
unless an exception is raised before exiting.
"""
def __init__(self, path, new=False):
mode = 'w+' if new else 'r+'
super().__init__(path, mode)
self.config = None
self.new = new
def __enter__(self) -> AttrOrderedDict:
if not self.new:
self.config = json.load(self.fd, object_pairs_hook=AttrOrderedDict)
else:
self.config = AttrOrderedDict()
return self.config
def __exit__(self, exc_type, exc_value, traceback):
# Only write the new value out if there wasn't an exception
if not exc_type:
self.fd.seek(0)
json.dump(self.config, self.fd, allow_nan=False, indent='\t')
self.fd.truncate()
super().__exit__(exc_type, exc_value, traceback)

View File

@ -1,52 +0,0 @@
"""
Dictionary classes used to represent JSON config files in memory.
"""
from collections import OrderedDict
from amanuensis.errors import ReadOnlyError
class AttrOrderedDict(OrderedDict):
"""
An OrderedDict with attribute access to known keys and explicit
creation of new keys.
"""
def __getattr__(self, key):
if key not in self:
raise AttributeError(key)
return self[key]
def __setattr__(self, key, value):
if key not in self:
raise AttributeError(key)
self[key] = value
def new(self, key, value):
"""Setter for adding new keys"""
if key in self:
raise KeyError("Key already exists: '{}'".format(key))
self[key] = value
class ReadOnlyOrderedDict(OrderedDict):
"""
An OrderedDict that cannot be modified with attribute access to
known keys.
"""
def __readonly__(self, *args, **kwargs):
raise ReadOnlyError("Cannot modify a ReadOnlyOrderedDict")
def __init__(self, *args, **kwargs):
super(ReadOnlyOrderedDict, self).__init__(*args, **kwargs)
self.__setitem__ = self.__readonly__
self.__delitem__ = self.__readonly__
self.pop = self.__readonly__
self.popitem = self.__readonly__
self.clear = self.__readonly__
self.update = self.__readonly__
self.setdefault = self.__readonly__
def __getattr__(self, key):
if key not in self:
raise AttributeError(key)
return self[key]

View File

@ -1,160 +0,0 @@
"""
Config directory abstractions that encapsulate path munging and context
manager usage.
"""
import os
import re
from typing import Iterable
from amanuensis.errors import MissingConfigError, ConfigAlreadyExistsError
from .context import json_ro, json_rw
def is_guid(s: str) -> bool:
return bool(re.match(r'[0-9a-z]{32}', s.lower()))
class ConfigDirectoryContext():
"""
Base class for CRUD operations on config files in a config
directory.
"""
def __init__(self, path: str):
self.path: str = path
if not os.path.isdir(self.path):
raise MissingConfigError(path)
def new(self, filename) -> json_rw:
"""
Creates a JSON file that doesn't already exist.
"""
if not filename.endswith('.json'):
filename = f'{filename}.json'
fpath: str = os.path.join(self.path, filename)
if os.path.isfile(fpath):
raise ConfigAlreadyExistsError(fpath)
return json_rw(fpath, new=True)
def read(self, filename) -> json_ro:
"""
Loads a JSON file in read-only mode.
"""
if not filename.endswith('.json'):
filename = f'{filename}.json'
fpath: str = os.path.join(self.path, filename)
if not os.path.isfile(fpath):
raise MissingConfigError(fpath)
return json_ro(fpath)
def edit(self, filename, create=False) -> json_rw:
"""
Loads a JSON file in write mode.
"""
if not filename.endswith('.json'):
filename = f'{filename}.json'
fpath: str = os.path.join(self.path, filename)
if not create and not os.path.isfile(fpath):
raise MissingConfigError(fpath)
return json_rw(fpath, new=create)
def delete(self, filename) -> None:
"""Deletes a file."""
if not filename.endswith('.json'):
filename = f'{filename}.json'
fpath: str = os.path.join(self.path, filename)
if not os.path.isfile(fpath):
raise MissingConfigError(fpath)
os.remove(fpath)
def ls(self) -> Iterable[str]:
"""Lists all files in this directory."""
filenames: Iterable[str] = os.listdir(self.path)
return filenames
class ConfigFileConfigDirectoryContext(ConfigDirectoryContext):
"""
Config directory with a `config.json`.
"""
def __init__(self, path: str):
super().__init__(path)
config_path = os.path.join(self.path, 'config.json')
if not os.path.isfile(config_path):
raise MissingConfigError(config_path)
def edit_config(self) -> json_rw:
"""rw context manager for this object's config file."""
return self.edit('config')
def read_config(self) -> json_ro:
"""ro context manager for this object's config file."""
return self.read('config')
class IndexDirectoryContext(ConfigDirectoryContext):
"""
A lookup layer for getting config directory contexts for lexicon
or user directories.
"""
def __init__(self, path: str, cdc_type: type):
super().__init__(path)
index_path = os.path.join(self.path, 'index.json')
if not os.path.isfile(index_path):
raise MissingConfigError(index_path)
self.cdc_type = cdc_type
def __getitem__(self, key: str) -> ConfigFileConfigDirectoryContext:
"""
Returns a context to the given item. key is treated as the
item's id if it's a guid string, otherwise it's treated as
the item's indexed name and run through the index first.
"""
if not is_guid(key):
with self.read_index() as index:
iid = index.get(key)
if not iid:
raise MissingConfigError(key)
key = iid
return self.cdc_type(os.path.join(self.path, key))
def edit_index(self) -> json_rw:
return self.edit('index')
def read_index(self) -> json_ro:
return self.read('index')
class RootConfigDirectoryContext(ConfigFileConfigDirectoryContext):
"""
Context for the config directory with links to the lexicon and
user contexts.
"""
def __init__(self, path):
super().__init__(path)
self.lexicon: IndexDirectoryContext = IndexDirectoryContext(
os.path.join(self.path, 'lexicon'),
LexiconConfigDirectoryContext)
self.user: IndexDirectoryContext = IndexDirectoryContext(
os.path.join(self.path, 'user'),
UserConfigDirectoryContext)
class LexiconConfigDirectoryContext(ConfigFileConfigDirectoryContext):
"""
A config context for a lexicon's config directory.
"""
def __init__(self, path):
super().__init__(path)
self.draft: ConfigDirectoryContext = ConfigDirectoryContext(
os.path.join(self.path, 'draft'))
self.src: ConfigDirectoryContext = ConfigDirectoryContext(
os.path.join(self.path, 'src'))
self.article: ConfigDirectoryContext = ConfigDirectoryContext(
os.path.join(self.path, 'article'))
class UserConfigDirectoryContext(ConfigFileConfigDirectoryContext):
"""
A config context for a user's config directory.
"""

View File

@ -1,96 +0,0 @@
# Standard library imports
from collections import OrderedDict
import fcntl
import json
import os
import shutil
# Module imports
from amanuensis.resources import get_stream
from .context import json_ro, json_rw
def create_config_dir(config_dir, refresh=False):
"""
Create or refresh a config directory
"""
def prepend(*path):
joined = os.path.join(*path)
if not joined.startswith(config_dir):
joined = os.path.join(config_dir, joined)
return joined
# Create the directory if it doesn't exist.
if not os.path.isdir(config_dir):
os.mkdir(config_dir)
# The directory should be empty if we're not updating an existing one.
if len(os.listdir(config_dir)) > 0 and not refresh:
print("Directory {} is not empty".format(config_dir))
return -1
# Update or create global config.
def_cfg = get_stream("global.json")
global_config_path = prepend("config.json")
if refresh and os.path.isfile(global_config_path):
# We need to write an entirely different ordereddict to the config
# file, so we mimic the config.context functionality manually.
with open(global_config_path, 'r+', encoding='utf8') as cfg_file:
fcntl.lockf(cfg_file, fcntl.LOCK_EX)
old_cfg = json.load(cfg_file, object_pairs_hook=OrderedDict)
new_cfg = json.load(def_cfg, object_pairs_hook=OrderedDict)
merged = {}
for key in new_cfg:
merged[key] = old_cfg[key] if key in old_cfg else new_cfg[key]
if key not in old_cfg:
print("Added key '{}' to config".format(key))
for key in old_cfg:
if key not in new_cfg:
print("Config contains unknown key '{}'".format(key))
merged[key] = old_cfg[key]
cfg_file.seek(0)
json.dump(merged, cfg_file, allow_nan=False, indent='\t')
cfg_file.truncate()
fcntl.lockf(cfg_file, fcntl.LOCK_UN)
else:
with open(prepend("config.json"), 'wb') as f:
f.write(def_cfg.read())
# Ensure lexicon subdir exists.
if not os.path.isdir(prepend("lexicon")):
os.mkdir(prepend("lexicon"))
if not os.path.isfile(prepend("lexicon", "index.json")):
with open(prepend("lexicon", "index.json"), 'w') as f:
json.dump({}, f)
# Ensure user subdir exists.
if not os.path.isdir(prepend("user")):
os.mkdir(prepend("user"))
if not os.path.isfile(prepend('user', 'index.json')):
with open(prepend('user', 'index.json'), 'w') as f:
json.dump({}, f)
if refresh:
for dir_name in ('lexicon', 'user'):
# Clean up unindexed folders
with json_ro(prepend(dir_name, 'index.json')) as index:
known = list(index.values())
entries = os.listdir(prepend(dir_name))
for dir_entry in entries:
if dir_entry == "index.json":
continue
if dir_entry in known:
continue
print("Removing unindexed folder: '{}/{}'"
.format(dir_name, dir_entry))
shutil.rmtree(prepend(dir_name, dir_entry))
# Remove orphaned index listings
with json_rw(prepend(dir_name, 'index.json')) as index:
for name, entry in index.items():
if not os.path.isdir(prepend(dir_name, entry)):
print("Removing stale {} index entry '{}: {}'"
.format(dir_name, name, entry))
del index[name]