PK!encapsia_cli/__init__.pyPK!_h&encapsia_cli/add_superuser.py"""Create a superuser e.g. for bootstrapping.""" import re import click from encapsia_api import EncapsiaApi from encapsia_cli import lib # 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 @click.command() @click.argument("email", callback=validate_email) @click.argument("first_name") @click.argument("last_name") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def main(email, first_name, last_name, host, token): """Create superuser with given name and email. In addition to adding the given user, this will also add a Superuser role. """ api = EncapsiaApi(host, lib.get_env_var(token)) api.post( "roles", json=[ {"name": "Superuser", "alias": "Superuser", "capabilities": ["superuser"]} ], ) api.post( "users", json=[ { "email": email, "first_name": first_name, "last_name": last_name, "role": "Superuser", "enabled": True, "is_site_user": False, } ], ) PK!n,` ` encapsia_cli/config.py"""Get/set server configuration.""" import click from encapsia_api import EncapsiaApi from encapsia_cli import lib @click.group() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) @click.option( "--format", type=click.Choice(["json", "toml"]), default="toml", help="Format as JSON or TOML (default TOML)", ) @click.pass_context def app(ctx, host, token, format): """Get/set server configuration (not *trial* configuration).""" if "." not in host: host = host + ".encapsia.com" token = lib.get_env_var(token) ctx.obj["host"] = host ctx.obj["token"] = token ctx.obj["format"] = format @app.command() @click.pass_context def show(ctx): """Show entire configuration.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) lib.pretty_print(api.get_all_config(), ctx.obj["format"]) @app.command() @click.argument("output", type=click.File("w")) @click.pass_context def save(ctx, output): """Save entire configuration to given file.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) lib.pretty_print(api.get_all_config(), ctx.obj["format"], output=output) @app.command() @click.argument("input", type=click.File("r")) @click.pass_context def load(ctx, input): """Load (merge) configuration from given file.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) data = lib.parse(input.read(), ctx.obj["format"]) api.set_config_multi(data) @app.command() @click.argument("key") @click.pass_context def get(ctx, key): """Retrieve value against given key.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) lib.pretty_print(api.get_config(key), ctx.obj["format"]) @app.command() @click.argument("key") @click.argument("value") @click.pass_context def set(ctx, key, value): """Store value against given key.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) value = lib.parse(value, ctx.obj["format"]) api.set_config(key, value) @app.command() @click.argument("key") @click.pass_context def delete(ctx, key): """Delete value against given key.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) api.delete_config(key) def main(): app(obj={}) PK!a<encapsia_cli/dbctl.py"""Encapsia Database control actions e.g. backups, fixtures.""" import os import os.path import sys import time import click from encapsia_api import EncapsiaApi from encapsia_cli import lib def visual_poll(message, poll, NoTaskResultYet, wait=0.5): out = sys.stdout out.write(message) out.flush() result = poll() while result is NoTaskResultYet: time.sleep(wait) out.write(".") out.flush() result = poll() out.write("Done\n") out.flush() return result def dbctl_action(host, token, name, params, message): api = EncapsiaApi(host, token) poll, NoTaskResultYet = api.dbctl_action(name, params) result = visual_poll(message, poll, NoTaskResultYet) if result["status"] != "ok": raise click.Abort() return result["result"] @click.group() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) @click.pass_context def main(ctx, host, token): """Low-level Encapsia Database control.""" ctx.obj = dict(host=host, token=lib.get_env_var(token)) @main.command() @click.pass_context def list_fixtures(ctx): """List available fixtures.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "list_fixtures", dict(), "Fetching list of fixtures...", ) ) @main.command() @click.argument("name") @click.pass_context def create_fixture(ctx, name): """Create new fixture with given name.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "create_fixture", dict(name=name), "Create fixture {}...".format(name), ) ) @main.command() @click.argument("name") @click.pass_context def use_fixture(ctx, name): """Switch to fixture with given name.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "use_fixture", dict(name=name), "Switching to fixture {}...".format(name), ) ) @main.command() @click.argument("name") @click.pass_context def delete_fixture(ctx, name): """Delete fixture with given name.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "delete_fixture", dict(name=name), "Deleting fixture {}...".format(name), ) ) @main.command() @click.argument("name") @click.pass_context def create_extension_schema(ctx, name): """Create extension schema.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "create_extension_schema", dict(name=name), "Creating schema {}...".format(name), ) ) @main.command() @click.argument("name") @click.pass_context def delete_extension_schema(ctx, name): """Delete extension schema.""" print( dbctl_action( ctx.obj["host"], ctx.obj["token"], "delete_extension_schema", dict(name=name), "Deleting schema {}...".format(name), ) ) @main.command() @click.argument("name") @click.argument("filename") @click.pass_context def load_extension_sql(ctx, name, filename): """Load SQL from given file into extension schema.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) handle = api.dbctl_upload_data(filename) dbctl_action( ctx.obj["host"], ctx.obj["token"], "load_extension_sql", dict(name=name, data_handle=handle), "Loading SQL into schema {}...".format(name), ) @main.command() @click.argument("handle") @click.option( "--filename", default=None, help="Optional filename into which the data will be downloaded.", ) @click.pass_context def download_data(ctx, handle, filename): """Download data of given handle.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) temp_filename = api.dbctl_download_data(handle) if filename is None: filename = temp_filename else: os.rename(temp_filename, filename) size = os.path.getsize(filename) print("Downloaded {} bytes to {}".format(size, filename)) @main.command() @click.argument("filename") @click.pass_context def upload_data(ctx, filename): """Upload data in given file, printing a handle for re-use.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) handle = api.dbctl_upload_data(filename) size = os.path.getsize(filename) print("Uploaded {} bytes from {}".format(size, filename)) print("Handle: {}".format(handle)) return handle @main.command() @click.option( "--filename", default=None, help="Optional filename into which the data will be downloaded.", ) @click.pass_context def backup(ctx, filename): """Backup database to given filename (or temp one if not given).""" handle = dbctl_action( ctx.obj["host"], ctx.obj["token"], "backup_database", dict(), "Backing up database...", ) api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) temp_filename = api.dbctl_download_data(handle) if filename is None: filename = temp_filename else: os.rename(temp_filename, filename) size = os.path.getsize(filename) print("Downloaded {} bytes to {}".format(size, filename)) @main.command() @click.argument("filename") @click.pass_context def restore(ctx, filename): """Restore database from given backup file.""" api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) 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. api = EncapsiaApi(ctx.obj["host"], ctx.obj["token"]) poll, NoTaskResultYet = api.dbctl_action( "restore_database", dict(data_handle=handle) ) print("Database restore requested.") print("Please verify by other means (e.g. look at the logs).") PK!^-encapsia_cli/expire_token.py"""Expire Encapsia token from a server.""" import click from encapsia_api import EncapsiaApi from encapsia_cli import lib def expire_token(host, token): api = EncapsiaApi("https://{}".format(host), token) try: api.delete("logout") except IceApiError as e: lib.error("Failed to expire given token!") lib.error(str(e)) raise click.Abort() @click.command() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def main(host, token): """Expire token from server given by ENCAPSIA_HOST. The token itself should be in an environment variable (default ENCAPSIA_TOKEN). """ if "." not in host: host += ".encapsia.com" expire_token(host, lib.get_env_var(token)) PK!,N# # encapsia_cli/lib.pyimport contextlib import datetime import io import json import os import shutil import subprocess import tarfile import tempfile import time import click import toml from encapsia_api import EncapsiaApi def error(message): click.secho(message, fg="red") def log(message, nl=True): click.secho(message, fg="yellow", nl=nl) def log_output(message): click.secho(message, fg="green") def get_env_var(name): try: return os.environ[name] except KeyError: error("Environment variable {} does not exist!".format(name)) raise click.Abort() 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 directory finally: shutil.rmtree(directory) def run(*args, **kwargs): """Run external command.""" return subprocess.check_output(args, stderr=subprocess.STDOUT, **kwargs) def create_targz(directory, filename): with tarfile.open(filename, "w:gz") as tar: tar.add(directory, arcname=os.path.basename(directory)) def create_targz_as_bytes(directory): data = io.BytesIO() with tarfile.open(mode="w:gz", fileobj=data) as tar: tar.add(directory, arcname=os.path.basename(directory)) return data.getvalue() 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: click.echo(formatted) else: output.write(formatted) def parse(obj, format): if format == "json": return json.loads(obj) elif format == "toml": return toml.load(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_plugins_task(host, token, name, params, message, data=None): api = EncapsiaApi(host, token) poll, NoTaskResultYet = api.run_task( "pluginsmanager", "icepluginsmanager.{}".format(name), params, data ) result = visual_poll(message, poll, NoTaskResultYet) log_output(result["output"].strip()) if result["status"] != "ok": raise click.Abort() PK!m'#'#encapsia_cli/plugins.py"""Create and install plugins.""" import datetime import glob import os import shutil import sys import tempfile import click import toml from encapsia_api import EncapsiaApi from encapsia_cli import lib @click.group() def main(): """Create and install plugins.""" @main.command("dev-create-namespace") @click.argument("namespace") @click.argument("n_task_workers", default=1) @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def dev_create_namespace(namespace, n_task_workers, host, token): """Create namespace of given name. Only useful during developmment.""" lib.run_plugins_task( host, lib.get_env_var(token), "dev_create_namespace", dict(namespace=namespace, n_task_workers=n_task_workers), "Creating namespace", ) @main.command("dev-destroy-namespace") @click.argument("namespace") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def dev_destroy_namespace(namespace, host, token): """Destory namespace of given name. Only useful during development""" lib.run_plugins_task( host, lib.get_env_var(token), "dev_destroy_namespace", dict(namespace=namespace), "Destroying namespace", ) @main.command() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def info(host, token): """Provide some information about installed plugins.""" lib.run_plugins_task( host, lib.get_env_var(token), "list_namespaces", dict(), "Fetching list of namespaces", ) 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, ) with open(filename, "w") as f: toml.dump(obj, f) @main.command("create-from-webapp") @click.argument("webapp") @click.argument("version") @click.option("--email", prompt="Your email") def create_from_webapp(webapp, version, email): """Convert a webapp on S3 in to a plugin.""" lib.log("Fetching webapp {} version {}...".format(webapp, version)) with lib.temp_directory() as temp_directory: base_name = "plugin-{}-{}".format(webapp, version) base_dir = os.path.join(temp_directory, base_name) os.makedirs(base_dir) # Download everything from S3 into the webfiles folder. files_directory = os.path.join(base_dir, "webfiles") os.makedirs(files_directory) lib.run( "aws", "s3", "cp", "s3://ice-webapp-builds/{}/{}".format(webapp, version), files_directory, "--recursive", ) # Move out the views if they exist. views_directory = os.path.join(files_directory, "views") if os.path.exists(views_directory): shutil.move(views_directory, base_dir) # Move out the tasks if they exist. tasks_directory = os.path.join(files_directory, "tasks") if os.path.exists(tasks_directory): shutil.move(tasks_directory, base_dir) # Create plugin.toml plugin_filename = os.path.join(base_dir, "plugin.toml") make_plugin_toml_file( plugin_filename, webapp, "Webapp {}".format(webapp), version, email ) # Convert all into tar.gz filename = base_name + ".tar.gz" lib.create_targz(base_dir, filename) lib.log("Created plugin: {}".format(filename)) return filename @main.command() @click.argument("filename") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def install(filename, host, token): """Install plugin from given tar.gz file or directory.""" temp_filename = None if os.path.isdir(filename): fd, temp_filename = tempfile.mkstemp(suffix=".tar.gz") os.close(fd) lib.create_targz(filename, temp_filename) filename = temp_filename api = EncapsiaApi(host, lib.get_env_var(token)) blob_id = api.upload_file_as_blob(filename) # TODO create plugin entity # TODO also assign a blobtag? lib.log(f"Uploaded {filename} to blob: {blob_id}") lib.run_plugins_task( host, lib.get_env_var(token), "install_plugin", dict(blob_id=blob_id), "Installing", ) if temp_filename: os.remove(temp_filename) @main.command() @click.argument("namespace") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def uninstall(namespace, host, token): """Uninstall named plugin.""" lib.run_plugins_task( host, lib.get_env_var(token), "uninstall_plugin", dict(namespace=namespace), "Uninstalling {}".format(namespace), ) class LastUploadedVsModifiedTracker: DIRECTORIES = ["tasks", "views", "wheels", "webfiles", "schedules"] def __init__(self, directory, reset=False): self.directory = directory self.filename = os.path.join(self.directory, ".encapsia", "last_uploaded_plugin_parts.toml") if reset: self.make_empty() else: self.load() def make_empty(self): os.makedirs(os.path.join(self.directory, ".encapsia"), exist_ok=True) self.data = {} self.save() def load(self): if not os.path.exists(self.filename): self.make_empty() else: with open(self.filename) as f: self.data = toml.load(f) def save(self): with open(self.filename, "w") as f: toml.dump(self.data, f) def get_modified_directories(self): for name in self.DIRECTORIES: possible_files = glob.glob(os.path.join(self.directory, name, "**")) if possible_files: if name in self.data: last_changed_file = max(possible_files, key=os.path.getctime) last_modified = datetime.datetime.fromtimestamp(os.path.getctime(last_changed_file)) if last_modified > self.data[name]: yield name self.data[name] = datetime.datetime.utcnow() else: yield 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.option("--force", is_flag=True, help="Force an update of all parts of the plugin") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST)") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def dev_update(directory, force, host, token): """Update plugin parts which have changed since previous update. Optionally pass in the DIRECTORY of the plugin (defaults to cwd). """ if not os.path.exists(os.path.join(directory, "plugin.toml")): lib.error("Not in a plugin directory.") sys.exit(1) modified_plugin_directories = get_modified_plugin_directories(directory, reset=force) if modified_plugin_directories: with lib.temp_directory() as tmp_directory: shutil.copy(os.path.join(directory, "plugin.toml"), tmp_directory) for modified_directory in modified_plugin_directories: lib.log(f"Including: {modified_directory}") shutil.copytree(os.path.join(directory, modified_directory), os.path.join(tmp_directory, modified_directory)) lib.run_plugins_task( host, lib.get_env_var(token), "dev_update_plugin", dict(), "Uploading to server", data=lib.create_targz_as_bytes(tmp_directory), ) else: lib.log("Nothing to do.") PK! o encapsia_cli/schedule.py"""Manage task schedules.""" import click from encapsia_cli import lib @click.group() def main(): """Manage task schedules.""" @main.command("list") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def list_tasks(host, token): """List all scheduled tasks.""" lib.run_plugins_task( host, lib.get_env_var(token), "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.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def add_task( description, task_host, task_token, namespace, task, params, cron, jitter, host, token, ): """Add new scheduled task.""" lib.run_plugins_task( host, lib.get_env_var(token), "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("remove_in_namespace") @click.argument("namespace") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def remove_tasks_in_namespace(namespace, host, token): """Remove all scheduled tasks in given namespace.""" lib.run_plugins_task( host, lib.get_env_var(token), "remove_scheduled_tasks_in_namespace", dict(namespace=namespace), "Removing scheduled tasks", ) @main.command("remove") @click.argument("scheduled_task_id") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def remove_task(scheduled_task_id, host, token): """Remove scheduled task by id.""" lib.run_plugins_task( host, lib.get_env_var(token), "remove_scheduled_task", dict(scheduled_task_id=scheduled_task_id), "Removing scheduled tasks", ) PK!${m==encapsia_cli/system_user.py"""Manage system users (create, list, delete etc).""" import click from encapsia_api import EncapsiaApi from encapsia_cli import lib @click.group() def main(): """Manage system users.""" @main.command() @click.argument("description") @click.argument("capabilities") @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def add(description, capabilities, host, token): """Create system user with suitable user and role details. Use quoting in the shell appropriately. For example: encapsia-system-user add "This is a description" "capability1, capability2" """ api = EncapsiaApi(host, lib.get_env_var(token)) api.add_system_user(description, [x.strip() for x in capabilities.split(",")]) @main.command() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def show(host, token): """Print system users.""" api = EncapsiaApi(host, lib.get_env_var(token)) for su in api.get_system_users(): print(su) PK!'encapsia_cli/task.py"""Run an arbitrary Encapsia task.""" import click from encapsia_api import EncapsiaApi from encapsia_cli import lib @click.command() @click.argument("namespace") @click.argument("function") @click.argument("args", nargs=-1) @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) def main(host, token, namespace, function, args): """Run an arbitrary task in given plugin NAMESPACE and qualified FUNCTION e.g. encapsia-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 task arguments are encoded over a URL string). """ params = {} for arg in args: left, right = arg.split("=", 1) params[left.strip()] = right.strip() api = EncapsiaApi(host, lib.get_env_var(token)) poll, NoTaskResultYet = api.run_task(namespace, function, params) result = lib.visual_poll(f"Running task {namespace}", poll, NoTaskResultYet) lib.log_output(str(result)) PK!**encapsia_cli/whoami.py"""Print information about owner of token.""" import click from encapsia_api import EncapsiaApi from encapsia_cli import lib @click.command() @click.option("--host", envvar="ENCAPSIA_HOST", help="DNS name of Encapsia host (or ENCAPSIA_HOST).") @click.option( "--token", default="ENCAPSIA_TOKEN", help="Environment variable containing server token (default ENCAPSIA_TOKEN)", ) @click.option( "--format", type=click.Choice(["json", "toml"]), default="json", help="Format as JSON or TOML (default JSON)", ) def main(host, token, format): """Print information about current owner of token. The token itself should be in an environment variable (default ENCAPSIA_TOKEN). """ api = EncapsiaApi(host, lib.get_env_var(token)) lib.pretty_print(api.whoami(), format) PK!H:-encapsia_cli-0.1.0.dist-info/entry_points.txtmM }0$ƐvR@t~ϰlbPɔ _ (VqM+c>YGeҟlyFu^MqmcHQ8c-z[HǎVO.Ac|[u^|1pKhUU1r>iU=Uy,sMQ΁w'x;73Crsot,';^ Ӆ`ԡ]sZ-aֶ"OR+j eˎJ4~)^N^q8ʁwHfxI&Aƃi&Gπ=b-{Lp0?',6a-0hoZbK&sg= BRpwVW0۸XW&@.N0`Qe ` (1jl |R 1kpeG`"SZ3٘;cOC@= -dh}-zǰ1I:_PWծS6 9W4xNGYv\l&S[?&WM0;cdx%TQ9h ݐpQ|p٭)oPK!HA'z#encapsia_cli-0.1.0.dist-info/RECORD˖<y?  HqQJ'Y)`' kPI4PRI\Y$yen[(P` u<+kx}x.2؜j~'nh ׳CaچvS@zU|[Saz;MmIr='fSISς*W j~KAY3[vp\P_R AVC:7^^m*ko^5?vLUU_=d>TxZЁY0ތ!O-W.2\o.Mui(m#vamS ,[֒pӷLD"RVcG7:AW3k:p>z`bT*29!-c䖠9{N,'oLXQ9y#^hBxz@ųC< oI#7dI͓o~ =q݃2 7ܼΪ,L+q;<~я:8]TOn^571*2Ic)(g6ދe9!htn2z*F_PK!encapsia_cli/__init__.pyPK!_h&6encapsia_cli/add_superuser.pyPK!n,` ` Yencapsia_cli/config.pyPK!a<encapsia_cli/dbctl.pyPK!^-(encapsia_cli/expire_token.pyPK!,N# # ,encapsia_cli/lib.pyPK!m'#'#6encapsia_cli/plugins.pyPK! o OZencapsia_cli/schedule.pyPK!${m==egencapsia_cli/system_user.pyPK!'lencapsia_cli/task.pyPK!**qencapsia_cli/whoami.pyPK!H:-)uencapsia_cli-0.1.0.dist-info/entry_points.txtPK!kBړ66$vencapsia_cli-0.1.0.dist-info/LICENSEPK!HڽTU"zencapsia_cli-0.1.0.dist-info/WHEELPK!H/$jT^Z%{encapsia_cli-0.1.0.dist-info/METADATAPK!HA'z#}encapsia_cli-0.1.0.dist-info/RECORDPK