PK!click_skeleton/__init__.pyPK!wWWclick_skeleton/__main__.pyimport sys if __name__ == "__main__": from .skel import main sys.exit(main()) PK!.\::click_skeleton/config.pyimport click import asyncio import os import coloredlogs import logging import attr import pwd from pathlib import Path project = 'skeleton' logger = logging.getLogger(__name__) home = str(Path.home()) current_user = pwd.getpwuid(os.getuid()).pw_name DEFAULT_LOG = '{}/{}.log'.format(home, project) MB_LOG = 'MB_LOG' MB_DEBUG = 'MB_DEBUG' MB_TIMINGS = 'MB_TIMINGS' MB_VERBOSITY = 'MB_VERBOSITY' MB_DRY = 'MB_DRY' MB_QUIET = 'MB_QUIET' MB_NO_COLORS = 'MB_NO_COLORS' DEFAULT_VERBOSITY = 'warning' DEFAULT_DRY = False DEFAULT_QUIET = False DEFAULT_DEBUG = False DEFAULT_NO_COLORS = False DEFAULT_TIMINGS = False verbosities = {'debug': logging.DEBUG, 'info': logging.INFO, 'warning': logging.WARNING, 'error': logging.ERROR, 'critical': logging.CRITICAL} def check_file_writable(fnm): if os.path.exists(fnm): # path exists if os.path.isfile(fnm): return os.access(fnm, os.W_OK) else: return False pdir = os.path.dirname(fnm) if not pdir: pdir = '.' return os.access(pdir, os.W_OK) options = [ click.option('--log', help='Log file path', type=click.Path(), envvar=MB_LOG, default=DEFAULT_LOG, show_default=True), click.option('--debug', help='Be very verbose, same as --verbosity debug + hide progress bars', envvar=MB_DEBUG, default=DEFAULT_DEBUG, is_flag=True), click.option('--timings', help='Set verbosity to info and show execution timings', envvar=MB_TIMINGS, default=DEFAULT_TIMINGS, is_flag=True), click.option('--verbosity', help='Verbosity levels', envvar=MB_VERBOSITY, default=DEFAULT_VERBOSITY, type=click.Choice(verbosities.keys()), show_default=True), click.option('--dry', help='Take no real action', envvar=MB_DRY, default=DEFAULT_DRY, is_flag=True), click.option('--quiet', help='Disable progress bars', envvar=MB_QUIET, default=DEFAULT_QUIET, is_flag=True), click.option('--no-colors', help='Disable colorized output', envvar=MB_NO_COLORS, default=DEFAULT_NO_COLORS, is_flag=True), ] @attr.s class Config: log = attr.ib(default=DEFAULT_LOG) quiet = attr.ib(default=DEFAULT_QUIET) debug = attr.ib(default=DEFAULT_DEBUG) timings = attr.ib(default=DEFAULT_TIMINGS) dry = attr.ib(default=DEFAULT_DRY) no_colors = attr.ib(default=DEFAULT_NO_COLORS) verbosity = attr.ib(default=DEFAULT_VERBOSITY) fmt = attr.ib(default="%(asctime)s %(name)s[%(process)d] %(levelname)s %(message)s") def set(self, debug=None, timings=None, dry=None, quiet=None, verbosity=None, no_colors=None, log=None): from .helpers import str2bool self.log = log if log is not None else os.getenv(MB_LOG, str(DEFAULT_LOG)) self.quiet = quiet if quiet is not None else str2bool(os.getenv(MB_QUIET, str(DEFAULT_QUIET))) self.debug = debug if debug is not None else str2bool(os.getenv(MB_DEBUG, str(DEFAULT_DEBUG))) self.timings = timings if timings is not None else str2bool(os.getenv(MB_TIMINGS, str(DEFAULT_TIMINGS))) self.dry = dry if dry is not None else str2bool(os.getenv(MB_DRY, str(DEFAULT_DRY))) self.no_colors = no_colors if no_colors is not None else str2bool(os.getenv(MB_NO_COLORS, str(DEFAULT_NO_COLORS))) self.verbosity = verbosity if verbosity is not None else os.getenv(MB_VERBOSITY, DEFAULT_VERBOSITY) if self.timings: self.verbosity = 'info' self.quiet = True if self.debug: self.verbosity = 'debug' self.quiet = True if self.verbosity == 'debug': loop = asyncio.get_event_loop() loop.set_debug(True) # triggers : "'NoneType' object has no attribute 'co_filename' # loop.slow_callback_duration = 0.001 loop.slow_callback_duration = 0.005 self.level = verbosities[self.verbosity] logging.basicConfig(level=self.level, format=self.fmt) if not self.no_colors: coloredlogs.install(level=self.level, fmt=self.fmt) if self.log is not None: if check_file_writable(self.log): fh = logging.FileHandler(self.log) fh.setLevel(logging.DEBUG) logging.getLogger().addHandler(fh) else: logger.warning('No permission to write to %s for current user %s', self.log, current_user) logger.debug(self) def __repr__(self): fmt = 'log:{} timings:{} debug:{} quiet:{} dry:{} verbosity:{} no_colors:{}' return fmt.format(self.log, self.timings, self.debug, self.quiet, self.dry, self.verbosity, self.no_colors) config = Config() PK!SzMMclick_skeleton/helpers.pyimport asyncio import time import uvloop import click import logging import string import random import os import sys import humanfriendly from click_didyoumean import DYMGroup from functools import wraps from timeit import default_timer as timer from .config import config logger = logging.getLogger(__name__) asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) def random_password(size=8): alphabet = string.ascii_letters + string.digits return ''.join(random.choice(alphabet) for i in range(size)) class GroupWithHelp(DYMGroup): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @click.command('help') @click.argument('command', nargs=-1) @click.pass_context def _help(ctx, command): '''Print help''' if command: argument = command[0] c = self.get_command(ctx, argument) print(c.get_help(ctx)) else: print(ctx.parent.get_help()) self.add_command(_help) async def process(f, *args, **params): if asyncio.iscoroutinefunction(f): return await f(*args, **params) return f(*args, **params) def timeit(f): @wraps(f) async def wrapper(*args, **params): start = time.time() result = await process(f, *args, **params) for_human = seconds_to_human(time.time() - start) if config.timings: logger.info('TIMINGS %s: %s', f.__name__, for_human) return result return wrapper def add_options(options): def _add_options(func): for option in reversed(options): func = option(func) return func return _add_options def coro(f): f = asyncio.coroutine(f) @wraps(f) def wrapper(*args, **kwargs): loop = asyncio.get_event_loop() return loop.run_until_complete(f(*args, **kwargs)) return wrapper def drier(f): @wraps(f) async def wrapper(*args, **kwargs): if config.dry: args = [str(a) for a in args] + ["%s=%s" % (k, v) for (k, v) in kwargs.items()] logger.info('DRY RUN: %s(%s)', f.__name__, ','.join(args)) await asyncio.sleep(0) else: return await process(f, *args, **kwargs) return wrapper def str2bool(val): val = val.lower() if val in ('y', 'yes', 't', 'true', 'on', '1'): return 1 if val in ('n', 'no', 'f', 'false', 'off', '0'): return 0 raise ValueError("invalid truth value %r" % (val,)) def bytes_to_human(b): return humanfriendly.format_size(b) def seconds_to_human(s): import datetime return str(datetime.timedelta(seconds=s)) class Benchmark: def __init__(self, msg): self.msg = msg def __enter__(self): self.start = timer() return self def __exit__(self, *args): t = timer() - self.start logger.info("%s : %0.3g seconds", self.msg, t) self.time = t class LazyProperty: def __init__(self, fget): self.fget = fget self.func_name = fget.__name__ def __get__(self, obj, cls): if obj is None: return None value = self.fget(obj) setattr(obj, self.func_name, value) return value def raise_limits(): import resource try: _, hard = resource.getrlimit(resource.RLIMIT_NOFILE) logger.info("Current limits, soft and hard : %s %s", _, hard) resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard)) return True except Exception as e: logger.critical('You may need to check ulimit parameter: %s', e) return False def restart(): python = sys.executable print('Restarting myself: {} {}'.format(python, sys.argv)) # only works on linux, not windows with WSL # os.execl(python, python, * sys.argv) # permit to exit from a thread os._exit(0) PK!wwclick_skeleton/skel.py#!/usr/bin/env python3 import os import logging import click import click_completion import click_repl import attrdict from . import helpers from .config import project, config, options if os.path.islink(__file__): myself = os.readlink(__file__) else: myself = __file__ bin_folder = os.path.dirname(myself) commands_folder = 'commands' plugin_folder = os.path.join(bin_folder, commands_folder) CONTEXT_SETTINGS = {'auto_envvar_prefix': project.upper(), 'help_option_names': ['-h', '--help']} logger = logging.getLogger(project) def custom_startswith(string, incomplete): """A custom completion matching that supports case insensitive matching""" if os.environ.get('_{}_CASE_INSENSITIVE_COMPLETE'.format(project.upper())): string = string.lower() incomplete = incomplete.lower() return string.startswith(incomplete) click_completion.startswith = custom_startswith click_completion.init() class SubCommandLineInterface(helpers.GroupWithHelp): def list_commands(self, ctx): rv = [] if os.path.isdir(plugin_folder): for filename in os.listdir(plugin_folder): if filename.endswith('.py') and '__init__' not in filename: rv.append(filename[:-3]) all_commands = rv + super().list_commands(ctx) all_commands.sort() return all_commands def get_command(self, ctx, name): ns = {} fn = os.path.join(plugin_folder, name + '.py') try: with open(fn) as f: code = compile(f.read(), fn, 'exec') ns['__name__'] = '{}.{}'.format(commands_folder, name) eval(code, ns, ns) except FileNotFoundError: return super().get_command(ctx, name) return ns['cli'] @click.group(cls=SubCommandLineInterface, context_settings=CONTEXT_SETTINGS) @click.version_option("1.0") @helpers.add_options(options) @click.pass_context def cli(ctx, **kwargs): """click-skeleton""" ctx.obj = attrdict.AttrDict ctx.obj.folder = bin_folder config.set(**kwargs) ctx.obj.config = config def main(): click_repl.register_repl(cli) cli() PK!Ha1;/click_skeleton-0.1.1.dist-info/entry_points.txtN+I/N.,()JL-NI-ϳsa\=*713 PK!H+dUT$click_skeleton-0.1.1.dist-info/WHEEL HM K-*ϳR03rOK-J,/R(O-)T0343 /, (-JLR()*M IL*4KM̫PK!HX'click_skeleton-0.1.1.dist-info/METADATA=O0=F+NTV[*%D |Jcv;0ٺ{ίuqĈP3^thHp,Ñ4EgOe6I.3i w ez{]W𫰝WdO]xYy eJY7],vCPXhh@y,ߚzߩ`~TCͺfUEuהEpy!;^e9i禌Ғ9kt~PK!HIֈo%click_skeleton-0.1.1.dist-info/RECORD;@|~ 8L[e|.lh(w7- ;.,1$'D?5L;)'_B7,-U/e"VXkhpK*aiQb o0f_*?l7<#VGBz*7ƹW9#-^AΟRN/[q:dV#hg'_al;c%#Ë lQ?EVrCpkeMX2y8.bf.'