amanuensis/cli/__init__.py

@@ -1,94 +1,72 @@
-from argparse import ArgumentParser
+#
-import logging
+# The cli module must not import other parts of the application at the module
-import logging.config
+# level. This is because most other modules depend on the config module. The
-
+# config module may depend on __main__'s commandline parsing to locate config
-import amanuensis.cli.admin
+# files, and __main__'s commandline parsing requires importing (but not
+# executing) the functions in the cli module. Thus, cli functions must only
+# import the config module inside the various command methods, which are only
+# run after commandline parsing has already occurred.
+#
 
 
-LOGGING_CONFIG = {
+def server_commands(commands={}):
-    "version": 1,
+	if commands:
-    "formatters": {
+		return commands
-        "fmt_basic": {
+	import amanuensis.cli.server
-            "validate": True,
+	for name, func in vars(amanuensis.cli.server).items():
-            "format": "%(message)s",
+		if name.startswith("command_"):
-        },
+			name = name[8:].replace("_", "-")
-        "fmt_detailed": {
+			commands[name] = func
-            "validate": True,
+	return commands
-            "format": "%(asctime)s %(levelname)s %(message)s"
-        },
-    },
-    "handlers": {
-        "hnd_stderr": {
-            "class": "logging.StreamHandler",
-            "level": "INFO",
-            "formatter": "fmt_basic",
-        },
-    },
-    "loggers": {
-        __name__: {
-            "level": "DEBUG",
-            "handlers": ["hnd_stderr"]
-        }
-    },
-}
 
 
-def add_subcommand(subparsers, module) -> None:
+def lexicon_commands(commands={}):
-    """Add a cli submodule's commands as a subparser."""
+	if commands:
-    # Get the command information from the module
+		return commands
-    command_name: str = getattr(module, "COMMAND_NAME")
+	import amanuensis.cli.lexicon
-    command_help: str = getattr(module, "COMMAND_HELP")
+	for name, func in vars(amanuensis.cli.lexicon).items():
-    if not command_name and command_help:
+		if name.startswith("command_"):
-        return
+			name = name[8:].replace("_", "-")
-
+			commands["lexicon-" + name] = func
-    # Add the subparser for the command and set a default action
+	return commands
-    command_parser: ArgumentParser = subparsers.add_parser(
-        command_name, help=command_help
-    )
-    command_parser.set_defaults(func=lambda args: command_parser.print_usage())
-
-    # Add all subcommands in the command module
-    subcommands = command_parser.add_subparsers(metavar="SUBCOMMAND")
-    for name, obj in vars(module).items():
-        if name.startswith("command_"):
-            # Hyphenate subcommand names
-            sc_name: str = name[8:].replace("_", "-")
-            # Only the first line of the subcommand function docstring is used
-            sc_help = ((obj.__doc__ or "").strip() or "\n").splitlines()[0]
-
-            # Add the command and any arguments defined by its decorators
-            subcommand: ArgumentParser = subcommands.add_parser(
-                sc_name, help=sc_help, description=obj.__doc__
-            )
-            subcommand.set_defaults(func=obj)
-            for args, kwargs in obj.__dict__.get("add_argument", []):
-                subcommand.add_argument(*args, **kwargs)
 
 
-def init_logger(args):
+def user_commands(commands={}):
-    """Set up logging based on verbosity args"""
+	if commands:
-    if (args.verbose):
+		return commands
-        handler = LOGGING_CONFIG["handlers"]["hnd_stderr"]
+	import amanuensis.cli.user
-        handler["formatter"] = "fmt_detailed"
+	for name, func in vars(amanuensis.cli.user).items():
-        handler["level"] = "DEBUG"
+		if name.startswith("command_"):
-    logging.config.dictConfig(LOGGING_CONFIG)
+			name = name[8:].replace("_", "-")
+			commands["user-" + name] = func
+	return commands
 
 
-def main():
+def get_commands():
-    """CLI entry point"""
+	return {**server_commands(), **lexicon_commands(), **user_commands()}
-    # Set up the top-level parser
-    parser = ArgumentParser()
-    parser.set_defaults(
-        parser=parser,
-        func=lambda args: parser.print_usage(),
-    )
-    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
 
-    # Add commands from cli submodules
-    subparsers = parser.add_subparsers(metavar="COMMAND")
-    add_subcommand(subparsers, amanuensis.cli.admin)
 
-    # Parse args and execute the desired action
+def cmd_desc(func):
-    args = parser.parse_args()
+	return ((func.__doc__ or "").strip() or '\n').splitlines()[0]
-    init_logger(args)
+
-    args.func(args)
+
+def describe_commands():
+	longest = max(map(len, server_commands().keys()))
+	server_desc = "General commands:\n{}\n".format("\n".join([
+		" {1:<{0}} : {2}".format(longest, name, cmd_desc(func))
+		for name, func in server_commands().items()
+	]))
+
+	longest = max(map(len, lexicon_commands().keys()))
+	lexicon_desc = "Lexicon commands:\n{}\n".format("\n".join([
+		" {1:<{0}} : {2}".format(longest, name, cmd_desc(func))
+		for name, func in lexicon_commands().items()
+	]))
+
+	longest = max(map(len, user_commands().keys()))
+	user_desc = "User commands:\n{}\n".format("\n".join([
+		" {1:<{0}} : {2}".format(longest, name, cmd_desc(func))
+		for name, func in user_commands().items()
+	]))
+
+	return "\n".join([server_desc, lexicon_desc, user_desc])

amanuensis/cli/admin.py

@@ -1,58 +0,0 @@
-import collections
-import json
-import logging
-import os
-
-from amanuensis.db import DbContext
-
-from .helpers import add_argument
-
-
-COMMAND_NAME = "admin"
-COMMAND_HELP = "Interact with Amanuensis."
-
-LOG = logging.getLogger(__name__)
-
-
-@add_argument("path", metavar="DB_PATH", help="Path to where the database should be created")
-@add_argument("--force", "-f", action="store_true", help="Overwrite existing database")
-@add_argument("--verbose", "-v", action="store_true", help="Enable db echo")
-def command_init_db(args) -> int:
-    """
-    Initialize the Amanuensis database.
-    """
-    # Check if force is required
-    if not args.force and os.path.exists(args.path):
-        args.parser.error(f"{args.path} already exists and --force was not specified")
-
-    # Initialize the database
-    db_uri = f"sqlite:///{os.path.abspath(args.path)}"
-    LOG.info(f"Creating database at {db_uri}")
-    db = DbContext(db_uri, debug=args.verbose)
-    db.create_all()
-
-    LOG.info("Done")
-    return 0
-
-
-@add_argument("path", metavar="CONFIG_PATH", help="Path to the config file")
-def command_secret_key(args) -> int:
-    """
-    Generate a Flask secret key.
-
-    The Flask server will not run unless a secret key has
-    been generated.
-    """
-    # Load the json config
-    with open(args.path, mode="r", encoding="utf8") as f:
-        config = json.load(f, object_pairs_hook=collections.OrderedDict)
-
-    # Set the secret key to a new random string
-    config["SECRET_KEY"] = os.urandom(32).hex()
-
-    # Write the config back out
-    with open(args.path, mode="w", encoding="utf8") as f:
-        json.dump(config, f, indent=2)
-
-    LOG.info("Regenerated Flask secret key")
-    return 0

amanuensis/cli/helpers.py

@@ -1,20 +1,209 @@
-"""
+# Standard library imports
-Helpers for cli commands.
+from argparse import ArgumentParser
-"""
+from functools import wraps
+from json.decoder import JSONDecodeError
+from logging import getLogger
+from sys import exc_info
+
+logger = getLogger(__name__)
+
+
+#
+# The add_argument and no_argument function wrappers allow the same
+# function to both configure a command and execute it. This keeps
+# command argument configuration close to where the command is defined
+# and reduces the number of things the main parser has to handle.
+#
 
 
 def add_argument(*args, **kwargs):
-    """Defines an argument to a cli command."""
+	"""Passes the given args and kwargs to subparser.add_argument"""
 
-    def argument_adder(command_func):
+	def argument_adder(command):
-        """Decorator function for storing parser args on the function."""
+		@wraps(command)
+		def augmented_command(cmd_args):
+			# Add this wrapper's command in the parser pass
+			if isinstance(cmd_args, ArgumentParser):
+				cmd_args.add_argument(*args, **kwargs)
+				# If there are more command wrappers, pass through to them
+				if command.__dict__.get('wrapper', False):
+					command(cmd_args)
+				# Parser pass doesn't return a value
+				return None
 
-        # Store the kw/args in the function dictionary
+			# Pass through transparently in the execute pass
-        add_args = command_func.__dict__.get("add_argument", [])
+			return command(cmd_args)
-        add_args.append((args, kwargs))
-        command_func.__dict__["add_argument"] = add_args
 
-        # Return the same function
+		# Mark the command as wrapped so control passes through
-        return command_func
+		augmented_command.__dict__['wrapper'] = True
+		return augmented_command
 
-    return argument_adder
+	return argument_adder
+
+
+def no_argument(command):
+	"""Noops for subparsers"""
+	@wraps(command)
+	def augmented_command(cmd_args):
+		# Noop in the parser pass
+		if isinstance(cmd_args, ArgumentParser):
+			return None
+		# Pass through in the execute pass
+		return command(cmd_args)
+
+	return augmented_command
+
+
+#
+# Many commands require specifying a lexicon or user to operate on, so
+# the requires_lexicon and requires_user wrappers replace @add_argument
+# as well as automatically create the model for the object from the
+# provided identifier.
+#
+
+
+LEXICON_ARGS = ['--lexicon']
+LEXICON_KWARGS = {
+	'metavar': 'LEXICON',
+	'dest': 'lexicon',
+	'help': 'Specify a user to operate on'}
+
+
+def requires_lexicon(command):
+	@wraps(command)
+	def augmented_command(cmd_args):
+		# Add lexicon argument in parser pass
+		if isinstance(cmd_args, ArgumentParser):
+			cmd_args.add_argument(*LEXICON_ARGS, **LEXICON_KWARGS)
+			# If there are more command wrappers, pass through to them
+			if command.__dict__.get('wrapper', False):
+				command(cmd_args)
+			# Parser pass doesn't return a value
+			return None
+
+		# Verify lexicon argument in execute pass
+		val = getattr(cmd_args, 'lexicon', None)
+		if not val:
+			logger.error("Missing --lexicon argument")
+			return -1
+		try:
+			model_factory = cmd_args.model_factory
+			cmd_args.lexicon = model_factory.lexicon(val)
+		except Exception:
+			ex_type, value, tb = exc_info()
+			logger.error(
+				f'Loading lexicon "{val}" failed with '
+				f'{ex_type.__name__}: {value}')
+			return -1
+		return command(cmd_args)
+
+	augmented_command.__dict__['wrapper'] = True
+	return augmented_command
+
+
+USER_ARGS = ['--user']
+USER_KWARGS = {
+	'metavar': 'USER',
+	'dest': 'user',
+	'help': 'Specify a user to operate on'}
+
+
+def requires_user(command):
+	@wraps(command)
+	def augmented_command(cmd_args):
+		# Add user argument in parser pass
+		if isinstance(cmd_args, ArgumentParser):
+			cmd_args.add_argument(*USER_ARGS, **USER_KWARGS)
+			# If there are more command wrappers, pass through to them
+			if command.__dict__.get('wrapper', False):
+				command(cmd_args)
+			# Parser pass doesn't return a value
+			return None
+
+		# Verify user argument in execute pass
+		val = getattr(cmd_args, "user", None)
+		if not val:
+			logger.error("Missing --user argument")
+			return -1
+		try:
+			model_factory = cmd_args.model_factory
+			cmd_args.user = model_factory.user(val)
+		except Exception:
+			ex_type, value, tb = exc_info()
+			logger.error(
+				f'Loading user "{val}" failed with '
+				f'{ex_type.__name__}: {value}')
+			return -1
+		return command(cmd_args)
+
+	augmented_command.__dict__['wrapper'] = True
+	return augmented_command
+
+
+# Wrapper for aliasing commands
+def alias(cmd_alias):
+	"""Adds an alias to the function dictionary"""
+	def aliaser(command):
+		aliases = command.__dict__.get('aliases', [])
+		aliases.append(cmd_alias)
+		command.__dict__['aliases'] = aliases
+		return command
+	return aliaser
+
+
+# Helpers for common command tasks
+
+CONFIG_GET_ROOT_VALUE = object()
+
+
+def config_get(cfg, pathspec):
+	"""
+	Performs config --get for a given config
+
+	cfg is from a `with json_ro` context
+	path is the full pathspec, unsplit
+	"""
+	import json
+
+	if pathspec is CONFIG_GET_ROOT_VALUE:
+		path = []
+	else:
+		path = pathspec.split(".")
+	for spec in path:
+		if spec not in cfg:
+			logger.error("Path not found: {}".format(pathspec))
+			return -1
+		cfg = cfg.get(spec)
+	print(json.dumps(cfg, indent=2))
+	return 0
+
+
+def config_set(obj_id, cfg, set_tuple):
+	"""
+	Performs config --set for a given config
+
+	config is from a "with json_rw" context
+	set_tuple is a tuple of the pathspec and the value
+	"""
+	import json
+	pathspec, value = set_tuple
+	if not pathspec:
+		logger.error("Path must be non-empty")
+	path = pathspec.split('.')
+	try:
+		value = json.loads(value)
+	except JSONDecodeError:
+		pass  # Leave value as string
+	for spec in path[:-1]:
+		if spec not in cfg:
+			logger.error("Path not found")
+			return -1
+		cfg = cfg.get(spec)
+	key = path[-1]
+	if key not in cfg:
+		logger.error("Path not found")
+		return -1
+	old_value = cfg[key]
+	cfg[key] = value
+	logger.info("{}.{}: {} -> {}".format(obj_id, pathspec, old_value, value))
+	return 0

amanuensis/cli/server.py

@@ -0,0 +1,120 @@
+import logging
+import os
+
+from amanuensis.config import RootConfigDirectoryContext
+
+from .helpers import (
+	add_argument,
+	no_argument,
+	alias,
+	config_get,
+	config_set,
+	CONFIG_GET_ROOT_VALUE)
+
+logger = logging.getLogger(__name__)
+
+
+@alias('i')
+@add_argument("--refresh",
+	action="store_true",
+	help="Refresh an existing config directory")
+def command_init(args):
+	"""
+	Initialize a config directory at --config-dir
+
+	A clean config directory will contain a config.json, a
+	lexicon config directory, and a user config directory.
+
+	Refreshing an existing directory will add keys to the global config that
+	are present in the default configs. Users and lexicons that are missing
+	from the indexes will be deleted, and stale index entries will be removed.
+	"""
+	# Module imports
+	from amanuensis.config.init import create_config_dir
+
+	# Verify arguments
+	if args.refresh and not os.path.isdir(args.config_dir):
+		print("Error: couldn't find directory '{}'".format(args.config_dir))
+
+	# Internal call
+	create_config_dir(args.config_dir, args.refresh)
+	logger.info(f'Initialized config dir at {args.config_dir}')
+	return 0
+
+
+@alias('gs')
+@no_argument
+def command_generate_secret(args):
+	"""
+	Generate a Flask secret key
+
+	The Flask server will not run unless a secret key has
+	been generated.
+	"""
+	root: RootConfigDirectoryContext = args.root
+	secret_key: bytes = os.urandom(32)
+	with root.edit_config() as cfg:
+		cfg.secret_key = secret_key.hex()
+	logger.info("Regenerated Flask secret key")
+	return 0
+
+
+@alias('r')
+@add_argument("-a", "--address", default="127.0.0.1")
+@add_argument("-p", "--port", default="5000")
+@add_argument("--debug", action="store_true")
+def command_run(args):
+	"""
+	Run the default Flask server
+
+	The default Flask server is not secure, and should
+	only be used for development.
+	"""
+	from amanuensis.server import get_app
+
+	root: RootConfigDirectoryContext = args.root
+
+	with root.read_config() as cfg:
+		if cfg.secret_key is None:
+			logger.error("Can't run server without a secret_key. "
+				"Run generate-secet first.")
+			return -1
+
+	get_app(root).run(host=args.address, port=args.port, debug=args.debug)
+	return 0
+
+
+@alias('n')
+@add_argument("--get",
+	metavar="PATHSPEC",
+	dest="get",
+	nargs="?",
+	const=CONFIG_GET_ROOT_VALUE,
+	help="Get the value of a config key")
+@add_argument("--set",
+	metavar=("PATHSPEC", "VALUE"),
+	dest="set",
+	nargs=2,
+	help="Set the value of a config key")
+def command_config(args):
+	"""
+	Interact with the global config
+
+	PATHSPEC is a path into the config object formatted as
+	a dot-separated sequence of keys.
+	"""
+	root: RootConfigDirectoryContext = args.root
+
+	if args.get and args.set:
+		logger.error("Specify one of --get and --set")
+		return -1
+
+	if args.get:
+		with root.read_config() as cfg:
+			config_get(cfg, args.get)
+
+	if args.set:
+		with root.edit_config() as cfg:
+			config_set("config", cfg, args.set)
+
+	return 0

amanuensis/log/__init__.py

@@ -0,0 +1,5 @@
+from .setup import init_logging
+
+__all__ = [member.__name__ for member in [
+	init_logging
+]]

amanuensis/log/setup.py

@@ -0,0 +1,45 @@
+import logging
+import logging.handlers
+
+
+basic_formatter = logging.Formatter(
+	fmt='[{levelname}] {message}',
+	style='{')
+detailed_formatter = logging.Formatter(
+	fmt='[{asctime}] [{levelname}:{filename}:{lineno}] {message}',
+	style='{')
+basic_console_handler = logging.StreamHandler()
+basic_console_handler.setLevel(logging.INFO)
+basic_console_handler.setFormatter(basic_formatter)
+detailed_console_handler = logging.StreamHandler()
+detailed_console_handler.setLevel(logging.DEBUG)
+detailed_console_handler.setFormatter(detailed_formatter)
+
+
+def get_file_handler(filename: str) -> logging.Handler:
+	handler = logging.handlers.RotatingFileHandler(
+		filename=filename,
+		maxBytes=1000000,
+		backupCount=10,
+		delay=True,
+		encoding='utf8',
+	)
+	handler.setLevel(logging.DEBUG)
+	handler.setFormatter(detailed_formatter)
+	return handler
+
+
+def init_logging(verbose: bool, log_filename: str):
+	"""
+	Initializes the Amanuensis logger settings
+	"""
+	logger = logging.getLogger("amanuensis")
+	if log_filename:
+		logger.addHandler(get_file_handler(log_filename))
+		logger.setLevel(logging.DEBUG)
+	elif verbose:
+		logger.addHandler(detailed_console_handler)
+		logger.setLevel(logging.DEBUG)
+	else:
+		logger.addHandler(basic_console_handler)
+		logger.setLevel(logging.INFO)

pyproject.toml

@@ -17,7 +17,6 @@ black = "^21.5b2"
 mypy = "^0.812"
 
 [tool.poetry.scripts]
-amanuensis-cli = "amanuensis.cli:main"
 amanuensis-server = "amanuensis.server:run"
 
 [tool.black]