Python invoke 模块，task() 实例源码

def clientWrapper(service, region=None):
    if region:
        cl = boto3.client(service, region_name=region)
    else:
        try:
            cl = boto3.client(service)
        except NoRegionError:
            print("You need to specify a region in some way. See "
                  "http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-"
                  "getting-started.html for ways that you can set it "
                  "globally, or pass 'region' as an argument to this task.")
            raise SystemExit(1)
    return cl


# Default namespace

def coverage(c, html=True):
    """
    Run coverage with coverage.py.
    """
    # NOTE: this MUST use coverage itself, and not pytest-cov, because the
    # latter is apparently unable to prevent pytest plugins from being loaded
    # before pytest-cov itself is able to start up coverage.py! The result is
    # that coverage _always_ skips over all module level code, i.e. constants,
    # 'def' lines, etc. Running coverage as the "outer" layer avoids this
    # problem, thus no need for pytest-cov.
    # NOTE: this does NOT hold true for NON-PYTEST code, so
    # pytest-relaxed-USING modules can happily use pytest-cov.
    c.run("coverage run --source=pytest_relaxed -m pytest")
    if html:
        c.run("coverage html")
        c.run("open htmlcov/index.html")


# TODO: good candidate for builtin-to-invoke "just wrap <other task> with a
# tiny bit of behavior", and/or args/kwargs style invocations

def deploy(ctx):
    pull_result = ctx.run("git pull")
    if is_task_file_changed(pull_result):
        sys.exit("Pyinvoke task file(s) is changed. Please re-run this task.")
    ctx.run("pip install -r requirements.txt")
    ctx.run("python manage.py makemigrations")
    ctx.run("python manage.py migrate")
    ctx.run("python manage.py collectstatic --noinput")
    # Reload application.
    ctx.run("touch ../reload")
    # Restart receiving server and batch server.
    ctx.run("sudo restart sh8recv", pty=True)
    ctx.run("sudo restart sh8batch", pty=True)
    # Give some time for reloading application.
    time.sleep(1)
    # Test if the site works well.
    with urlopen("https://sh8.email") as response:
        if not response.getcode() == 200:
            sys.exit("CRITICAL: The site respond CODE " + response.getcode())
    # Success!
    print("Deploy succeeded.")

def watcher(task, *args, **kwargs):
    while True:
        run('clear')
        kwargs['warn'] = True
        task(*args, **kwargs)
        try:
            run(
                'inotifywait -q -e create -e modify -e delete '
                '--exclude ".*\.(pyc|sw.)" -r spotifyconnect/ tests/')
        except KeyboardInterrupt:
            sys.exit()

def confirm(prompt='Continue?\n', failure_prompt='User cancelled task'):
    '''
    Prompt the user to continue. Repeat on unknown response. Raise
    ParseError on negative response
    '''
    response = input(prompt)

    try:
        response_bool = strtobool(response)
    except ValueError:
        print('Unkown Response. Confirm with y, yes, t, true, on or 1; cancel with n, no, f, false, off or 0.')
        confirm(prompt, failure_prompt)

    if not response_bool:
        raise ParseError(failure_prompt)

def task(*args, **kwargs):
    if not kwargs and len(args) == 1 and callable(args[0]):
        task = _task(args[0])
        ns.add_task(task)
    else:
        def decorator(func):
            task = _task(func)
            ns.add_task(task)
            return task
        return decorator

def clean_all(ctx, dry_run=False):
    """Clean up everything, even the precious stuff.
    NOTE: clean task is executed first.
    """
    cleanup_dirs(ctx.clean_all.directories or [], dry_run=dry_run)
    cleanup_dirs(ctx.clean_all.extra_directories or [], dry_run=dry_run)
    cleanup_files(ctx.clean_all.files or [], dry_run=dry_run)
    cleanup_files(ctx.clean_all.extra_files or [], dry_run=dry_run)
    execute_cleanup_tasks(ctx, cleanup_all_tasks, dry_run=dry_run)
    clean(ctx, dry_run=dry_run)

def clean_all(ctx, dry_run=False):
    """Clean up everything, even the precious stuff.
    NOTE: clean task is executed first.
    """
    cleanup_dirs(ctx.clean_all.directories or [], dry_run=dry_run)
    cleanup_dirs(ctx.clean_all.extra_directories or [], dry_run=dry_run)
    cleanup_files(ctx.clean_all.files or [], dry_run=dry_run)
    cleanup_files(ctx.clean_all.extra_files or [], dry_run=dry_run)
    execute_cleanup_tasks(ctx, cleanup_all_tasks, dry_run=dry_run)
    clean(ctx, dry_run=dry_run)

def list_tasks():
    """Show tasks, basically an alias for --list.

    Needed as a default task due to https://github.com/pyinvoke/invoke/issues/180
    """
    run('invoke --list')

def tasks(self):
        """Function that turns a collection of tasks
        suitable for pyinvoke_

        Example::

            from app.web import ac
            ns = Collection()
            ns.add_collection(ac.tasks())

        .. _pyinvoke: http://www.pyinvoke.org/
        """
        from invoke import task, Collection

        @task
        def list(ctx):
            """Show all clients in the database"""
            from json import dumps
            with (self.app or current_app).app_context():
                print(dumps([
                    dict(c) for c in self.client_class.all()
                ]))

        @task
        def show(ctx, clientKey):
            """Lookup one client from the database"""
            from json import dumps
            with (self.app or current_app).app_context():
                print(dumps(dict(self.client_class.load(clientKey))))

        @task
        def install(ctx, data):
            """Add a given client from the database"""
            from json import loads
            with (self.app or current_app).app_context():
                client = loads(data)
                self.client_class.save(client)
                print("Added")

        @task()
        def uninstall(ctx, clientKey):
            """Remove a given client from the database"""
            with (self.app or current_app).app_context():
                self.client_class.delete(clientKey)
                print("Deleted")

        ns = Collection('clients')
        ns.add_task(list)
        ns.add_task(show)
        ns.add_task(install)
        ns.add_task(uninstall)
        return ns

def rm(path, verbose=True):
    """
    Remove file or directory in path.
    """

    if os.path.exists(path):
        if verbose:
            print('removing', path, end=' ')

        if os.path.isdir(path):
            for base, files, dirs in os.walk(path):
                dirs_and_files = dirs + files
                for path in dirs_and_files:
                    path = os.path.join(base, path)
                    rm(path, verbose=False)
        else:
            os.unlink(path)

        if verbose:
            print('...done!')

# @task
# def lint(ctx):
#     """
#     Run the linter
#     """
#     print('not ready...')
#
#
# @task
# def diagnose(ctx):
#     print('not ready...')
#
#
# @task
# def publish(ctx):
#     print('not ready...')
#
#
# @task
# def release(ctx):
#     print('not ready...')
#
#
# @task
# def release_check(ctx):
#     print('not ready...')
#
#
# @task
# def http_serve(ctx):
#     print('not ready...')