PK!2ךsnakeless/__init__.pyimport sys from .cli_controller import SnakelessCli def main(argv=sys.argv[1:]): snakeless_cli = SnakelessCli() return snakeless_cli.run(argv) PK! <]]snakeless/__main__.pyimport sys from . import main if __name__ == '__main__': sys.exit(main(sys.argv[1:])) PK!v{WKKsnakeless/cli_controller.pyfrom cliff import help as cliff_help from cliff.app import App from cliff.commandmanager import CommandManager class SnakelessCli(App): def __init__(self): super().__init__( description="Snakeless CLI", version="0.1.2", command_manager=CommandManager("snakeless.cli"), deferred_help=True, ) def initialize_app(self, argv): if self.interactive_mode: # disable interactive mode action = cliff_help.HelpAction(None, None, default=self) action(self.parser, self.options, None, None) PK!˶CCsnakeless/commands/__init__.pyfrom .check import Check # noqa from .deploy import Deploy # noqaPK!{00snakeless/commands/check.pyimport logging import fs from cliff.command import Command from schema import SchemaError from ..constants import CURRENT_DIR from ..helpers import ( check_config_existence, snakeless_spinner, parse_config ) from ..exceptions import CommandFailure class Check(Command): "A health-check of your snakeless setup" logger = logging.getLogger(__name__) def check_config_existence(self, root_fs): with snakeless_spinner( text="Checking the config existence.", spinner="dots" ) as spinner: config_file_exists = check_config_existence(root_fs) if not config_file_exists: raise CommandFailure("Config was not found.") else: spinner.succeed("Config was found.") def validate_config(self, root_fs): with snakeless_spinner( text="Validating the config file.", spinner="dots" ) as spinner: try: parse_config(root_fs) except fs.errors.ResourceNotFound: raise CommandFailure("Config does not anymore exist.") except SchemaError as exc: raise CommandFailure("Config validation failed") else: spinner.succeed("Config is valid.") def take_action(self, parsed_args): with fs.open_fs(CURRENT_DIR) as root_fs: try: self.check_config_existence(root_fs) self.validate_config(root_fs) except Exception as exc: self.logger.exception(exc, exc_info=True) PK!fsnakeless/commands/deploy.pyimport logging import fs from cliff.command import Command from ..constants import CURRENT_DIR from ..mixins import DeployerMixin, ConfigLoaderMixin class Deploy(Command, DeployerMixin, ConfigLoaderMixin): "Deploy some functions or all of themn" logger = logging.getLogger(__name__) def get_parser(self, prog_name): parser = super().get_parser(prog_name) parser.add_argument( "-f", nargs="*", default=".", help="A list of functions' names" ) return parser def take_action(self, parsed_args): parsed_args = vars(parsed_args) functions_to_deploy = parsed_args.get('f', []) with fs.open_fs(CURRENT_DIR) as root_fs: try: config = self.load_config(root_fs) if not functions_to_deploy: functions_to_deploy = config['functions'].keys() self.deploy_functions(config, functions_to_deploy) except Exception as exc: self.logger.exception(exc, exc_info=True) PK!C&&snakeless/constants.pyimport os CURRENT_DIR = os.getcwd() PK!E**snakeless/exceptions.pyclass CommandFailure(Exception): pass PK!QU snakeless/helpers.pyfrom functools import lru_cache from pkg_resources import iter_entry_points from contextlib import contextmanager from halo import Halo from yaml import load from schema import Schema, And, Use, Optional from .exceptions import CommandFailure try: from yaml import CLoader as Loader except ImportError: from yaml import Loader @contextmanager def snakeless_spinner(*args, **kwargs): spinner = Halo(*args, **kwargs) spinner.start() try: yield spinner except CommandFailure as exc: spinner.fail(str(exc)) raise except Exception as exc: spinner.fail( spinner.text + "\n" + "Unexpected exception." ) raise finally: spinner.stop() def check_config_existence(root_fs, file_name="snakeless.yml"): return root_fs.exists(file_name) @lru_cache() def get_providers(): providers = {} for entry_point in iter_entry_points("snakeless.providers"): providers[entry_point.name] = entry_point.load() return providers @lru_cache() def get_schemas(): schemas = {} for entry_point in iter_entry_points("snakeless.schemas"): schemas[entry_point.name] = entry_point.load() return schemas def get_provider(provider_name, config): providers = get_providers() for entry_point in iter_entry_points("snakeless.providers"): providers[entry_point.name] = entry_point.load() return providers[provider_name](config) @lru_cache(maxsize=None) def get_schema(provider_name): PROVIDERS = get_providers() SUPPORTED_PROVIDERS = PROVIDERS.keys() BASE_SCHEMA = { "project": { "name": str, "provider": And( Use(str), lambda provider: provider in SUPPORTED_PROVIDERS, error="Unsupported provider", ), }, } schemas = get_schemas() provider_schema = schemas[provider_name] return merge(BASE_SCHEMA, provider_schema) def parse_config( root_fs, file_name="snakeless.yml" ): config_file_data = root_fs.gettext("snakeless.yml") raw_parsed_config = load(config_file_data, Loader=Loader) try: provider_name = raw_parsed_config["project"]["provider"] except KeyError: raise else: provider_schema = get_schema(provider_name) validator = Schema(provider_schema, ignore_extra_keys=True) parse_config = validator.validate(raw_parsed_config) # TODO: validate a few more fields mannualy return parse_config def merge(source, destination): for key, value in source.items(): if isinstance(value, dict): node = destination.setdefault(key, {}) merge(value, node) else: destination[key] = value return destination PK!8x{``snakeless/mixins/__init__.pyfrom .deployer import DeployerMixin # noqa from .config_loader import ConfigLoaderMixin # noqaPK!Cx!snakeless/mixins/config_loader.pyimport fs from schema import SchemaError from ..helpers import snakeless_spinner, parse_config from ..exceptions import CommandFailure class ConfigLoaderMixin(object): def load_config(self, root_fs): with snakeless_spinner( text="Loading the config file...", spinner="dots" ) as spinner: try: config = parse_config(root_fs) except fs.errors.ResourceNotFound: raise CommandFailure("Config does not exist.") except SchemaError as exc: raise CommandFailure("Config validation failed.") else: spinner.succeed("The config file was loaded.") return config PK!5snakeless/mixins/deployer.pyfrom ..helpers import snakeless_spinner, get_provider class DeployerMixin(object): def deploy_functions(self, config, functions_to_deploy): with snakeless_spinner( text="Deploying functions...", spinner="dots" ) as spinner: provider_name = config["project"]["provider"] provider = get_provider(provider_name, config) for function_to_deploy in functions_to_deploy: spinner.text = ( f"Deploying the { function_to_deploy } function..." ) provider.deploy_function(function_to_deploy) spinner.succeed("All functions were deployed!") PK!snakeless/providers/__init__.pyPK!4ɛpK>X;baP>PK!H"snakeless-0.2.4.dist-info/METADATAUn7}W>DwTH6 SvjFѻc.pu8[OZqeIY3rO|-GDbm-CqV~ 7^#'4HTgۋ_m h<5lweeѵލ6LgBC=\֏=VtnyZ`+C6 4$E%*T\|P9,wʼnDj/`<߮&p윷?O\.ml[e8|_0/6 * ^rރb*f9x2y\j%1wUsz|pGD# ku&=i ꟱ ˳1\KĚ&׽gq W+'ݺU-ǫj uM#;564%u`v& ^Z]-o_4b{)/pVw8xH%0c)$ZCN۾b}CkY 3[*AIHQ=p}yԝFeFx <{ȥ8ȺV)_aMT3C΋'Ұ YXpچ396߹ʽh3 93Zx+ke 7PK!H*l}  snakeless-0.2.4.dist-info/RECORD}ɲH}= B&aQ fPepC $ mGGupz/O?CT3˶DqpKHq2KE1/aǣZB4JQs؆ +Iֺpa8hc3撶 ~]"1"*40ZYi])lQ+MŗG5#E= GsFc ]i6>> o#:?0r1w5% Bk7V+.'iRx0*IGC#G23p^sV1<(j=`1)ns#r. {FN 8UXf .usGLj5 ǎ1Ũg8ӹ+LGP@d^Ԇ[_6?縃E/ }F8ɓ