PK!,:[[encapsia_cli/__init__.py#: Keep in sync with git tag and package version in pyproject.toml. __version__ = "0.1.10" PK!'4encapsia_cli/config.py"""Get/set server configuration.""" import click from encapsia_cli import lib main = lib.make_main(__doc__) @main.command() @click.pass_obj def show(obj): """Show entire configuration.""" api = lib.get_api(**obj) lib.pretty_print(api.get_all_config(), "json") @main.command() @click.argument("output", type=click.File("w")) @click.pass_obj def save(obj, output): """Save entire configuration to given file.""" api = lib.get_api(**obj) lib.pretty_print(api.get_all_config(), "json", output=output) @main.command() @click.argument("input", type=click.File("r")) @click.pass_obj def load(obj, input): """Load (merge) configuration from given file.""" api = lib.get_api(**obj) data = lib.parse(input.read(), "json") api.set_config_multi(data) @main.command() @click.argument("key") @click.pass_obj def get(obj, key): """Retrieve value against given key.""" api = lib.get_api(**obj) lib.pretty_print(api.get_config(key), "json") @main.command() @click.argument("key") @click.argument("value") @click.pass_obj def set(obj, key, value): """Store value against given key.""" api = lib.get_api(**obj) value = lib.parse(value, "json") api.set_config(key, value) @main.command() @click.argument("key") @click.pass_obj def delete(obj, key): """Delete value against given key.""" api = lib.get_api(**obj) api.delete_config(key) PK!50$$encapsia_cli/database.py"""Backups and Restore encapsia databases.""" from pathlib import Path import click from encapsia_cli import lib main = lib.make_main(__doc__) @main.command() @click.argument( "filename", type=click.Path(writable=True, readable=False), required=False ) @click.pass_obj def backup(obj, filename): """Backup database to given filename. (or create a temp one if not given).""" if filename: filename = Path(filename) api = lib.get_api(**obj) handle = lib.dbctl_action(api, "backup_database", dict(), "Backing up database") temp_filename = Path( api.dbctl_download_data(handle) ) # TODO remove Path() once encapsia_api switched over to pathlib if filename is None: filename = temp_filename else: temp_filename.rename(filename) lib.log(f"Downloaded {filename.stat().st_size} bytes to {filename}") @main.command() @click.argument("filename", type=click.Path(exists=True)) @click.option("--yes", is_flag=True, help="Don't prompt the user for confirmation.") @click.pass_obj def restore(obj, filename, yes): """Restore database from given backup file.""" filename = Path(filename) if not yes: click.confirm( f'Are you sure you want to restore the database from "{filename}"?', abort=True, ) api = lib.get_api(**obj) handle = api.dbctl_upload_data(filename) # On a restore, the server is temporarily stopped. # This means that attempts to use it will generate a 500 error when # Nginx tries to check the permission. # Further, the current token may no longer work. poll, NoTaskResultYet = api.dbctl_action( "restore_database", dict(data_handle=handle) ) lib.log("Database restore requested.") lib.log("Please verify by other means (e.g. look at the logs).") PK!encapsia_cli/encapsia.py"""CLI to talk to an encapsia host. The following steps are used to determine the server URL and token: \b If provided, use the --host option to reference an entry in ~/.encapsia/credentials.toml Else if set, use ENCAPSIA_HOST to reference an entry in ~/.encapsia/credentials.toml Else if set, use ENCAPSIA_URL and ENCAPSIA_TOKEN directly. Else abort. The tool will also abort if instructed to lookup in ~/.encapsia/credentials.toml but cannot find a correct entry. """ import click import click_completion import encapsia_cli.config import encapsia_cli.database import encapsia_cli.fixtures import encapsia_cli.help import encapsia_cli.httpie import encapsia_cli.plugins import encapsia_cli.run import encapsia_cli.schedule import encapsia_cli.shell import encapsia_cli.token import encapsia_cli.users import encapsia_cli.version from encapsia_cli import lib #: Initialise click completion. click_completion.init() COMMANDS = { "config": encapsia_cli.config.main, "database": encapsia_cli.database.main, "fixtures": encapsia_cli.fixtures.main, "help": encapsia_cli.help.main, "httpie": encapsia_cli.httpie.main, "plugins": encapsia_cli.plugins.main, "run": encapsia_cli.run.main, "schedule": encapsia_cli.schedule.main, "shell": encapsia_cli.shell.main, "token": encapsia_cli.token.main, "users": encapsia_cli.users.main, "version": encapsia_cli.version.main, } class EncapsiaCli(click.MultiCommand): def list_commands(self, ctx): return sorted(COMMANDS.keys()) def get_command(self, ctx, name): try: return COMMANDS[name] except KeyError: lib.log_error(ctx.get_help()) lib.log_error() raise click.UsageError(f"Unknown command {name}") main = EncapsiaCli(help=__doc__) PK!%<encapsia_cli/fixtures.py"""Manage database fixtures.""" import click from encapsia_cli import lib main = lib.make_main(__doc__) @main.command("list") @click.pass_obj def list_fixtures(obj): """List available fixtures.""" api = lib.get_api(**obj) lib.log_output( lib.dbctl_action(api, "list_fixtures", dict(), f"Fetching list of fixtures") ) @main.command("create") @click.argument("name") @click.pass_obj def create_fixture(obj, name): """Create new fixture with given name.""" api = lib.get_api(**obj) lib.log_output( lib.dbctl_action( api, "create_fixture", dict(name=name), f"Creating fixture {name}" ) ) @main.command("use") @click.argument("name") @click.option("--yes", is_flag=True, help="Don't prompt the user for confirmation.") @click.pass_obj def use_fixture(obj, name, yes): """Switch to fixture with given name.""" if not yes: click.confirm( f'Are you sure you want to change the database to fixture "{name}"?', abort=True, ) api = lib.get_api(**obj) poll, NoTaskResultYet = api.dbctl_action("use_fixture", dict(name=name)) lib.log(f"Requested change to fixture {name}.") lib.log("Please verify by other means (e.g. look at the logs).") @main.command("delete") @click.argument("name") @click.option("--yes", is_flag=True, help="Don't prompt the user for confirmation.") @click.pass_obj def delete_fixture(obj, name, yes): """Delete fixture with given name.""" if not yes: click.confirm(f'Are you sure you want to delete fixture "{name}"?', abort=True) api = lib.get_api(**obj) lib.log_output( lib.dbctl_action( api, "delete_fixture", dict(name=name), f"Deleting fixture {name}" ) ) PK!f7encapsia_cli/help.pyimport click from encapsia_cli import lib @click.command("help") @lib.colour_option @click.argument("command", required=False) @click.pass_context def main(ctx, colour, command): """Print longer help information about the CLI.""" ctx.color = {"always": True, "never": False, "auto": None}[colour] root_command = ctx.parent.command if command: lib.log(root_command.get_command(ctx, command).get_help(ctx)) else: lib.log(root_command.get_help(ctx)) lib.log() lib.log("Subcommands:") long_list = [] for name in root_command.list_commands(ctx): command = root_command.get_command(ctx, name) if isinstance(command, click.Group): for subname in command.list_commands(ctx): subcommand = command.get_command(ctx, subname) help_str = subcommand.get_short_help_str() long_list.append((name, subname, help_str)) width = max(len(name) + len(subname) for (name, subname, _) in long_list) for name, subname, help_str in long_list: left = f"{name} {subname}" left = left + " " * (width + 2 - len(left)) lib.log(f" {left} {help_str}") PK!)encapsia_cli/httpie.py"""Helper to use httpie with the URL and credentials passed in.""" import subprocess import click from encapsia_cli import lib main = lib.make_main(__doc__) @main.command("shell") @click.pass_obj def shell(obj): """Launch an httpie interactive shell with passed-in credentials.""" api = lib.get_api(**obj) argv = [ "http-prompt", api.url, f"Authorization: Bearer {api.token}", "Accept: application/json", ] subprocess.run(argv) PK!ֲDencapsia_cli/lib.pyimport contextlib import datetime import io import json import re import shutil import subprocess import tarfile import tempfile import time from pathlib import Path import click import encapsia_api import toml def log(message="", nl=True): click.secho(message, fg="yellow", nl=nl) def log_output(message=""): click.secho(message, fg="green") def log_error(message="", abort=False): click.secho(message, fg="red", err=True) if abort: raise click.Abort() def pretty_print(obj, format, output=None): if format == "json": formatted = json.dumps(obj, sort_keys=True, indent=4).strip() elif format == "toml": formatted = toml.dumps(obj) if output is None: log_output(formatted) else: output.write(formatted) def get_api(**obj): try: url, token = encapsia_api.discover_credentials(obj["host"]) except encapsia_api.EncapsiaApiError as e: log_error(str(e), abort=True) return encapsia_api.EncapsiaApi(url, token) def add_docstring(value): """Decorator to add a docstring to a function.""" def _doc(func): func.__doc__ = value return func return _doc colour_option = click.option( "--colour", type=click.Choice(["always", "never", "auto"]), default="auto", help="Control colour on stdout.", envvar="ENCAPSIA_COLOUR", ) host_option = click.option( "--host", help="Name to use to lookup credentials in .encapsia/credentials.toml", ) def make_main(docstring, for_plugins=False): if for_plugins: @click.group() @colour_option @host_option @click.option( "--plugins-cache-dir", type=click.Path(), default="~/.encapsia/plugins-cache", help="Name of directory used to cache plugins.", ) @click.option( "--force/--no-force", default=False, help="Always fetch/build/etc again." ) @click.pass_context @add_docstring(docstring) def main(ctx, colour, host, plugins_cache_dir, force): ctx.color = {"always": True, "never": False, "auto": None}[colour] plugins_cache_dir = Path(plugins_cache_dir).expanduser() plugins_cache_dir.mkdir(parents=True, exist_ok=True) ctx.obj = dict(host=host, plugins_cache_dir=plugins_cache_dir, force=force) else: @click.group() @colour_option @host_option @click.pass_context @add_docstring(docstring) def main(ctx, colour, host): ctx.color = {"always": True, "never": False, "auto": None}[colour] ctx.obj = dict(host=host) return main # See http://www.regular-expressions.info/email.html EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$") def validate_email(ctx, param, value): if not EMAIL_REGEX.match(value): raise click.BadParameter("Not a valid email address") return value def get_utc_now_as_iso8601(): return str(datetime.datetime.utcnow()) @contextlib.contextmanager def temp_directory(): """Context manager for creating a temporary directory. Cleans up afterwards. """ directory = tempfile.mkdtemp() try: yield Path(directory) finally: shutil.rmtree(directory) def most_recently_modified(directory): """Return datetime of most recently changed file in directory.""" files = list(directory.glob("**/*.*")) if files: return datetime.datetime.utcfromtimestamp(max(t.stat().st_mtime for t in files)) else: return None def run(*args, **kwargs): """Run external command.""" return subprocess.check_output(args, stderr=subprocess.STDOUT, **kwargs) def read_toml(filename): with filename.open() as f: return toml.load(f) def write_toml(filename, obj): with filename.open("w") as f: toml.dump(obj, f) def create_targz(directory, filename): with tarfile.open(filename, "w:gz") as tar: tar.add(directory, arcname=directory.name) def create_targz_as_bytes(directory): data = io.BytesIO() with tarfile.open(mode="w:gz", fileobj=data) as tar: tar.add(directory, arcname=directory.name) return data.getvalue() def parse(obj, format): if format == "json": return json.loads(obj) elif format == "toml": return toml.loads(obj) def visual_poll(message, poll, NoTaskResultYet, wait=0.2): log(message, nl=False) result = poll() count = 0 while result is NoTaskResultYet: time.sleep(wait) log(".", nl=False) count += 1 result = poll() if count < 3: log("." * (3 - count), nl=False) log("Done") return result def run_task(api, namespace, name, params, message, data=None): """Return the raw json result or log (HTTP) error and abort.""" poll, NoTaskResultYet = api.run_task(namespace, name, params, data) try: return visual_poll(message, poll, NoTaskResultYet) except encapsia_api.EncapsiaApiError as e: result = e.args[0] log_error(f"\nStatus: {result['status']}") log_error(result.get("exc_info"), abort=True) def run_plugins_task(api, name, params, message, data=None): """Log the result from pluginmanager, which will either be successful or not.""" reply = run_task( api, "pluginsmanager", "icepluginsmanager.{}".format(name), params, message, data, ) if reply["status"] == "ok": log(f"Status: {reply['status']}") log_output(reply["output"].strip()) else: log_error(str(reply), abort=True) def run_job(api, namespace, function, params, data=None): """Run job, wait for it to complete, and log all joblogs; or log error from the task.""" extra_headers = {"Content-type": "application/octet-stream"} if data else None reply = api.post( ("jobs", namespace, function), params=params, data=data, extra_headers=extra_headers, ) task_id = reply["result"]["task_id"] job_id = reply["result"]["job_id"] class NoResultYet: pass def get_task_result(): reply = api.get(("tasks", namespace, task_id)) rest_api_result = reply["result"] task_status = rest_api_result["status"] task_result = rest_api_result["result"] if task_status == "finished": return task_result elif task_status == "failed": log_error(f"\nStatus: {task_status}") log_error(rest_api_result.get("exc_info"), abort=True) else: return NoResultYet visual_poll("Running job", get_task_result, NoResultYet) reply = api.get(("jobs", namespace, job_id)) return reply["result"] def dbctl_action(api, name, params, message): poll, NoTaskResultYet = api.dbctl_action(name, params) result = visual_poll(message, poll, NoTaskResultYet) if result["status"] != "ok": raise click.Abort() return result["result"] PK!!*!*encapsia_cli/plugins.py"""Install, uninstall, create, and update plugins.""" import datetime import re import shutil import sys import urllib.request from pathlib import Path import click import toml from encapsia_cli import lib main = lib.make_main(__doc__, for_plugins=True) @main.command() @click.pass_obj def info(obj): """Provide some information about installed plugins.""" api = lib.get_api(**obj) lib.run_plugins_task(api, "list_namespaces", dict(), "Fetching list of namespaces") def read_toml(filename): with filename.open() as f: return toml.load(f) @main.command() @click.option("--versions", help="TOML file containing webapp names and versions.") @click.pass_obj def install(obj, versions): """Install plugins from version.toml file.""" plugins_cache_dir = obj["plugins_cache_dir"] versions = Path(versions) api = lib.get_api(**obj) for name, version in read_toml(versions).items(): plugin_filename = plugins_cache_dir / f"plugin-{name}-{version}.tar.gz" if not plugin_filename.exists(): lib.log_error( f"Unable to find plugin {name} version {name} in cache ({plugins_cache_dir})", abort=True, ) # TODO only upload if not already installed? (unless --force) blob_id = api.upload_file_as_blob(plugin_filename.as_posix()) # TODO create plugin entity and pass that in (the pluginsmanager creates the pluginlogs entity) lib.log(f"Uploaded {plugin_filename} to blob: {blob_id}") lib.run_plugins_task(api, "install_plugin", dict(blob_id=blob_id), "Installing") @main.command() @click.argument("namespace") @click.pass_obj def uninstall(obj, namespace): """Uninstall named plugin.""" if not obj["force"]: click.confirm( f'Are you sure you want to uninstall the plugin (delete all!) from namespace "{namespace}"?', abort=True, ) api = lib.get_api(**obj) lib.run_plugins_task( api, "uninstall_plugin", dict(namespace=namespace), f"Uninstalling {namespace}" ) class LastUploadedVsModifiedTracker: DIRECTORIES = ["tasks", "views", "wheels", "webfiles", "schedules"] def __init__(self, directory, reset=False): self.directory = directory encapsia_directory = directory / ".encapsia" encapsia_directory.mkdir(parents=True, exist_ok=True) self.filename = encapsia_directory / "last_uploaded_plugin_parts.toml" if reset: self.make_empty() else: self.load() def make_empty(self): self.data = {} self.save() def load(self): if not self.filename.exists(): self.make_empty() else: with self.filename.open() as f: self.data = toml.load(f) def save(self): with self.filename.open("w") as f: toml.dump(self.data, f) def get_modified_directories(self): for name in self.DIRECTORIES: last_modified = lib.most_recently_modified(self.directory / name) if last_modified is not None: if name in self.data: if last_modified > self.data[name]: yield Path(name) self.data[name] = datetime.datetime.utcnow() else: yield Path(name) self.data[name] = datetime.datetime.utcnow() self.save() def get_modified_plugin_directories(directory, reset=False): return list( LastUploadedVsModifiedTracker(directory, reset=reset).get_modified_directories() ) @main.command("dev-update") @click.argument("directory", default=".") @click.pass_obj def dev_update(obj, directory): """Update plugin parts which have changed since previous update. Optionally pass in the DIRECTORY of the plugin (defaults to cwd). """ directory = Path(directory) plugin_toml_path = directory / "plugin.toml" if not plugin_toml_path.exists(): lib.log_error("Not in a plugin directory.") sys.exit(1) modified_plugin_directories = get_modified_plugin_directories( directory, reset=obj["force"] ) if modified_plugin_directories: with lib.temp_directory() as temp_directory: shutil.copy(plugin_toml_path, temp_directory) for modified_directory in modified_plugin_directories: lib.log(f"Including: {modified_directory}") shutil.copytree( directory / modified_directory, temp_directory / modified_directory ) api = lib.get_api(**obj) lib.run_plugins_task( api, "dev_update_plugin", dict(), "Uploading to server", data=lib.create_targz_as_bytes(temp_directory), ) else: lib.log("Nothing to do.") @main.command("dev-create-namespace") @click.argument("namespace") @click.argument("n_task_workers", default=1) @click.pass_obj def dev_create_namespace(obj, namespace, n_task_workers): """Create namespace of given name. Only useful during developmment.""" api = lib.get_api(**obj) lib.run_plugins_task( api, "dev_create_namespace", dict(namespace=namespace, n_task_workers=n_task_workers), "Creating namespace", ) @main.command("dev-destroy-namespace") @click.argument("namespace") @click.pass_obj def dev_destroy_namespace(obj, namespace): """Destroy namespace of given name. Only useful during development""" api = lib.get_api(**obj) lib.run_plugins_task( api, "dev_destroy_namespace", dict(namespace=namespace), "Destroying namespace" ) def make_plugin_toml_file(filename, name, description, version, created_by): obj = dict( name=name, description=description, version=version, created_by=created_by, n_task_workers=1, reset_on_install=True, ) lib.write_toml(filename, obj) @main.command() @click.option( "--versions", type=click.Path(exists=True), help="TOML file containing webapp names and versions.", ) @click.option("--email", prompt="Your email", help="Email creator of the plugins.") @click.option( "--s3-directory", default="ice-webapp-builds", help="Base directory on S3." ) @click.pass_obj def build_from_legacy_s3(obj, versions, email, s3_directory): """Build plugins from legacy webapps hosted on AWS S3.""" plugins_cache_dir = obj["plugins_cache_dir"] force = obj["force"] versions = Path(versions) for name, version in lib.read_toml(versions).items(): output_filename = Path(plugins_cache_dir, f"plugin-{name}-{version}.tar.gz") if not force and output_filename.exists(): lib.log(f"Found: {output_filename} (Skipping)") else: _download_and_build_plugin_from_s3( s3_directory, name, version, email, output_filename ) lib.log(f"Created: {output_filename}") def _download_and_build_plugin_from_s3( s3_directory, name, version, email, output_filename ): with lib.temp_directory() as temp_directory: base_dir = temp_directory / f"plugin-{name}-{version}" base_dir.mkdir() # Download everything from S3 into the webfiles folder. # (we will move out the views and tasks if present). files_directory = base_dir / "webfiles" files_directory.mkdir() lib.run( "aws", "s3", "cp", f"s3://{s3_directory}/{name}/{version}", files_directory.as_posix(), "--recursive", ) # Move out the views if they exist. views_directory = files_directory / "views" if views_directory.exists(): views_directory.rename(base_dir / "views") # Move out the wheels if they exist. wheels_directory = files_directory / "tasks/wheels" if wheels_directory.exists(): wheels_directory.rename(base_dir / "wheels") # Move out the tasks if they exist. tasks_directory = files_directory / "tasks" if tasks_directory.exists(): tasks_directory.rename(base_dir / "tasks") # Create a plugin.toml manifest. make_plugin_toml_file( base_dir / "plugin.toml", name, f"Webapp {name}", version, email ) # Convert all into tar.gz lib.create_targz(base_dir, output_filename) @main.command() @click.argument("sources", nargs=-1) @click.pass_obj def build_from_src(obj, sources): """Build plugins from given source directories.""" plugins_cache_dir = obj["plugins_cache_dir"] force = obj["force"] for source_directory in sources: source_directory = Path(source_directory) manifest = read_toml(source_directory / "plugin.toml") name = manifest["name"] version = manifest["version"] output_filename = plugins_cache_dir / f"plugin-{name}-{version}.tar.gz" if not force and output_filename.exists(): lib.log(f"Found: {output_filename} (Skipping)") else: with lib.temp_directory() as temp_directory: base_dir = temp_directory / f"plugin-{name}-{version}" base_dir.mkdir() for t in ( "webfiles", "views", "tasks", "wheels", "schedules", "plugin.toml", ): source_t = source_directory / t if source_t.exists(): if source_t.is_file(): shutil.copy(source_t, base_dir / t) else: shutil.copytree(source_t, base_dir / t) lib.create_targz(base_dir, output_filename) lib.log(f"Created: {output_filename}") @main.command() @click.argument("url") @click.pass_obj def fetch_from_url(obj, url): """Copy a plugin from given URL into the plugin cache.""" plugins_cache_dir = obj["plugins_cache_dir"] force = obj["force"] full_name = url.rsplit("/", 1)[-1] m = re.match(r"plugin-([^-]*)-([^-]*).tar.gz", full_name) if m: output_filename = plugins_cache_dir / full_name if not force and output_filename.exists(): lib.log(f"Found: {output_filename} (Skipping)") else: filename, headers = urllib.request.urlretrieve(url) shutil.move(filename, output_filename) lib.log(f"Created: {output_filename}") else: lib.log_error("That doesn't look like a plugin. Aborting!", abort=True) PK!G0@@encapsia_cli/run.py"""Run an Encapsia task or view.""" import json import click from encapsia_cli import lib main = lib.make_main(__doc__) def _output(result, save_as): """Deal with result from task or view etc. Either print (pretty if possible) or write to file. """ if not isinstance(result, str): result = json.dumps(result) if save_as: save_as.write(result) lib.log(f"Saved result to {save_as.name}") else: try: # Try to pretty print if it converts to JSON. data = json.loads(result) lib.pretty_print(data, "json") except json.decoder.JSONDecodeError: # Otherwise print normally. lib.log_output(str(result)) @main.command("task") @click.argument("namespace") @click.argument("function") @click.argument("args", nargs=-1) @click.option( "--upload", type=click.File("rb"), help="Name of file to upload and hence pass to the task", ) @click.option( "--save-as", type=click.File("w"), help="Name of file in which to save result" ) @click.pass_obj def run_task(obj, namespace, function, args, upload, save_as): """Run a task in given plugin NAMESPACE and FUNCTION with ARGS. E.g. \b encapsia run task example_namespace test_module.test_function x=3 y=tim "z=hello stranger" Note that all args must be named and the values are all considered strings (not least because arguments are encoded over a URL string). """ api = lib.get_api(**obj) params = {} for arg in args: left, right = arg.split("=", 1) params[left.strip()] = right.strip() result = lib.run_task( api, namespace, function, params, f"Running task {namespace}", data=upload ) _output(result, save_as) @main.command("job") @click.argument("namespace") @click.argument("function") @click.argument("args", nargs=-1) @click.option( "--upload", type=click.File("rb"), help="Name of file to upload and hence pass to the job", ) @click.option( "--save-as", type=click.File("w"), help="Name of file in which to save result" ) @click.pass_obj def run_job(obj, namespace, function, args, upload, save_as): """Run a job in given plugin NAMESPACE and FUNCTION with ARGS. E.g. \b encapsia run job example_namespace test_module.test_function x=3 y=tim "z=hello stranger" Note that all args must be named and the values are all considered strings (not least because arguments are encoded over a URL string). """ api = lib.get_api(**obj) params = {} for arg in args: left, right = arg.split("=", 1) params[left.strip()] = right.strip() result = lib.run_job(api, namespace, function, params, data=upload) _output(result, save_as) @main.command("view") @click.argument("namespace") @click.argument("function") @click.argument("args", nargs=-1) @click.option( "--upload", type=click.File("rb"), help="Name of file to upload and hence pass to the task", ) @click.option( "--save-as", type=click.File("w"), help="Name of file in which to save result" ) @click.pass_obj def run_view(obj, namespace, function, args, upload, save_as): """Run a view in given plugin NAMESPACE and FUNCTION with ARGS. e.g. \b encapsia run view example_namespace test_view 3 tim limit=45 If an ARGS contains an "=" sign then send it as an optional query string argument. Otherwise send it as a URL path segment. """ # Split command line arguments into path segments and query string arguments. query_args = {} path_segments = [] for arg in args: if "=" in arg: left, right = arg.split("=", 1) query_args[left] = right else: path_segments.append(arg) extra_headers = None if upload: extra_headers = {"Content-Type": "text/plain"} api = lib.get_api(**obj) response = api.call_api( "GET", ["views", namespace, function] + path_segments, return_json=False, params=query_args, extra_headers=extra_headers, data=upload, ) _output(response.text, save_as) PK!+encapsia_cli/schedule.py"""Manage task schedules.""" import click from encapsia_cli import lib main = lib.make_main(__doc__) @main.command("list") @click.pass_obj def list_tasks(obj): """List all scheduled tasks.""" api = lib.get_api(**obj) lib.run_plugins_task( api, "list_scheduled_tasks", {}, "Fetching list of scheduled tasks" ) @main.command("add") @click.option("--description", prompt="Description", required=True) @click.option("--task-host", prompt="Task host", required=True) @click.option("--task-token", prompt="Task token", required=True) @click.option("--namespace", prompt="Namespace", required=True) @click.option("--task", prompt="Task (function)", required=True) @click.option("--params", prompt="Params (dict of args to function)", required=True) @click.option( "--cron", prompt="Cron string (e.g. '*/5 * * * *' means every 5 mins)", required=True, ) @click.option("--jitter", prompt="Jitter (int)", type=int, required=True) @click.pass_obj def add_task( obj, description, task_host, task_token, namespace, task, params, cron, jitter ): """Add new scheduled task.""" api = lib.get_api(**obj) lib.run_plugins_task( api, "add_scheduled_task", dict( description=description, host=task_host, token=task_token, namespace=namespace, task=task, params=params, cron=cron, jitter=jitter, ), "Adding scheduled task", ) @main.command() @click.argument("namespace") @click.pass_obj def remove_tasks_in_namespace(obj, namespace): """Remove all scheduled tasks in given namespace.""" api = lib.get_api(**obj) lib.run_plugins_task( api, "remove_scheduled_tasks_in_namespace", dict(namespace=namespace), "Removing scheduled tasks", ) @main.command() @click.argument("scheduled_task_id") @click.pass_obj def remove_task(obj, scheduled_task_id): """Remove scheduled task by id.""" api = lib.get_api(**obj) lib.run_plugins_task( api, "remove_scheduled_task", dict(scheduled_task_id=scheduled_task_id), "Removing scheduled tasks", ) PK! encapsia_cli/shell.pyimport os import click import click_shell @click.command("shell") @click.option( "--host", help="Name to use to lookup credentials in .encapsia/credentials.toml" ) @click.pass_context def main(ctx, host): """Start an interactive shell for running the encapsia commands. The --host option internally sets the ENCAPSIA_HOST environment variable, which subsequent commands will pick up if a --host option is not set again. """ if host: os.environ["ENCAPSIA_HOST"] = host prompt = f"encapsia {host}> " else: prompt = "encapsia> " shell = click_shell.make_click_shell( ctx.parent, prompt=prompt, intro="Starting interactive shell...\nType help for help!", ) shell.cmdloop() PK!g4YYencapsia_cli/token.py"""Do things with an encapsia token.""" import click from encapsia_api import CredentialsStore, EncapsiaApiError from encapsia_cli import lib main = lib.make_main(__doc__) @main.command() @click.pass_obj def expire(obj): """Expire token from server, and update encapsia credentials if used.""" api = lib.get_api(**obj) try: api.delete("logout") lib.log("Expired token on server.") except EncapsiaApiError as e: lib.log_error("Failed to expire given token!") lib.log_error(str(e), abort=True) if obj["host"]: CredentialsStore().remove(obj["host"]) lib.log("Removed entry from encapsia credentials file.") @main.command() @click.pass_obj def whoami(obj): """Print information about current owner of token.""" api = lib.get_api(**obj) lib.pretty_print(api.whoami(), "toml") PK!<m/ encapsia_cli/users.py"""Manage users, including superuser and system users.""" import click import tabulate from encapsia_cli import lib main = lib.make_main(__doc__) @main.command() @click.argument("description") @click.argument("capabilities") @click.pass_obj def add_systemuser(obj, description, capabilities): """Create system user with suitable user and role.""" api = lib.get_api(**obj) api.add_system_user(description, [x.strip() for x in capabilities.split(",")]) @main.command() @click.argument("email", callback=lib.validate_email) @click.argument("first_name") @click.argument("last_name") @click.pass_obj def add_superuser(obj, email, first_name, last_name): """Create superuser with suitable user and role.""" api = lib.get_api(**obj) api.add_super_user(email, first_name, last_name) @main.command("list") @click.option("--super-users/--no-super-users", default=False) @click.option("--system-users/--no-system-users", default=False) @click.option("--all-users/--no-all-users", default=False) @click.pass_obj def list_users(obj, super_users, system_users, all_users): """List out information about users.""" api = lib.get_api(**obj) if not (super_users or system_users or all_users): # If no specific type of user specified then assume all-users was intended. all_users = True if super_users: lib.log_output("[Super users]") users = api.get_super_users() headers = ["email", "first_name", "last_name"] lib.log_output( tabulate.tabulate( [[getattr(row, header) for header in headers] for row in users], headers=headers, ) ) lib.log_output() if system_users: lib.log_output("[System users]") users = api.get_system_users() headers = ["email", "description", "capabilities"] lib.log_output( tabulate.tabulate( [[getattr(row, header) for header in headers] for row in users], headers=headers, ) ) lib.log_output() if all_users: lib.log_output("[All users]") users = api.get_all_users() headers = [ "email", "first_name", "last_name", "role", "enabled", "is_site_user", ] lib.log_output( tabulate.tabulate( [[row[header] for header in headers] for row in users], headers=headers ) ) lib.log_output() @main.command("delete") @click.argument("email", callback=lib.validate_email) @click.pass_obj def delete_user(obj, email): """Delete user (but *do not* delete any related role).""" api = lib.get_api(**obj) api.delete_user(email) PK!Xڏ8encapsia_cli/version.pyimport click import encapsia_api import encapsia_cli from encapsia_cli import lib @click.command("version") @click.pass_context @lib.colour_option def main(ctx, colour): """Print version information about the CLI.""" ctx.color = {"always": True, "never": False, "auto": None}[colour] lib.log(f"Encapsia CLI version: {encapsia_cli.__version__}") lib.log(f"Encapsia API version: {encapsia_api.__version__}") PK!H|||,7.encapsia_cli-0.1.10.dist-info/entry_points.txtN+I/N.,()JKN,(L1s2`<..PK!kBړ66%encapsia_cli-0.1.10.dist-info/LICENSEMIT License Copyright (c) 2019 Timothy Corbett-Clark Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HڽTU#encapsia_cli-0.1.10.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H0 &encapsia_cli-0.1.10.dist-info/METADATAV]S7}_q'v +ېRhÄ N逬֮6&!$3y(30+~sxI:wzHjkhun6}5P~vV!8uc@Ig<階!(.cJe}yg_a7l'N8潏Xuq xW_6::^6Bԃ _Y &MY6.mF4Xg +]u0ic8FF ([2]itvlTvAfj0/EA"Hܑr!Y+rW"M) V0 :Ԫk7NtGumIaPRM8- ^YnPK!H 1@$encapsia_cli-0.1.10.dist-info/RECORDɒJK3x AiuBЃC}3nDEdγOU7}{aj|Mq TRƂ3ֶ$ -U +%FQ6*+W&КW9zb7(E~o@= Q+jAb-A7p:AW(Zθu_`09l, p49< ŴN~NQ_`  |`szY=ݿ蛗Rz1eqIu==HS`N6-%]l aL%۷.ek6x>e #l|;FI°k,nZ&Tze `wȃ ta(/xֲmX5SQ꣈i^mV(zͭE!cHSAuP|Q̒"?u`IRq O7UD51$||(,`<az_VJt5 3n$HF] jh"FX ;NĐt7T2GGR# wl4=3GdΚd,e~q2D~`wo-੣vŕ@w%y ϑ߸&Hzd?Z_X^oÁtX12WڄLWE&4*"1kwC@vyԯ;ty5_<2fՁFB\o_L%wC;~7󉐿]rTn?#iaWXy\:l[`׋xLfM=^(S뮬^y2^;1n-ba~˭6'p61jIY|(Ydޚ9[]&%1'S@?PK!,:[[encapsia_cli/__init__.pyPK!'4encapsia_cli/config.pyPK!50$$Fencapsia_cli/database.pyPK! encapsia_cli/encapsia.pyPK!%<encapsia_cli/fixtures.pyPK!f7encapsia_cli/help.pyPK!)!encapsia_cli/httpie.pyPK!ֲD6#encapsia_cli/lib.pyPK!!*!*?encapsia_cli/plugins.pyPK!G0@@Yiencapsia_cli/run.pyPK!+yencapsia_cli/schedule.pyPK! encapsia_cli/shell.pyPK!g4YY˅encapsia_cli/token.pyPK!<m/ Wencapsia_cli/users.pyPK!Xڏ8dencapsia_cli/version.pyPK!H|||,7.Bencapsia_cli-0.1.10.dist-info/entry_points.txtPK!kBړ66%encapsia_cli-0.1.10.dist-info/LICENSEPK!HڽTU#3encapsia_cli-0.1.10.dist-info/WHEELPK!H0 &țencapsia_cli-0.1.10.dist-info/METADATAPK!H 1@$ޠencapsia_cli-0.1.10.dist-info/RECORDPK