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

def run_terraform(ctx, command, opts_list):
    # Add state file location option
    tf_state_file = '{0}/{1}'.format(ctx.terraform.dir, ctx.terraform.state)
    option = '-state={0}'.format(tf_state_file)
    opts_list.append(option)

    # Build terraform command
    opts = ' '.join(opts_list)
    tf_dir = ctx.terraform.dir
    terraform = 'terraform {0} {1} {2}'.format(command, opts, tf_dir)

    result = run(terraform, hide=False, warn=True)
    if result.ok:
        list(ctx)
    else:
        print("\n" + "-> " + result.command + "\n")
        print(result)
        raise exceptions.Failure(result)

def deploy_docs():
    """
    Based on https://gist.github.com/domenic/ec8b0fc8ab45f39403dd
    """
    run('rm -rf ./site/')
    build_docs()
    with util.cd('./site/'):
        run('git init')
        run('echo ".*pyc" > .gitignore')
        run('git config user.name "Travis CI"')
        run('git config user.email "%s"' % os.environ['EMAIL'])
        run('git add .')
        run('git commit -m "Deploy to GitHub Pages"')
        run(
            'git push --force --quiet "https://{GH_TOKEN}@{GH_REF}" '
            'master:gh-pages > /dev/null 2>&1'.format(
                GH_TOKEN=os.environ['GH_TOKEN'],
                GH_REF=os.environ['GH_REF'],
            )
        )

def gunicorn(ctx, bind='localhost:8000', workers=13, collectstatic=True):
    """
    Starts Gunicorn server. This is not an optimal deployment, since you should
    serve static files with a proxy such as nginx. However, Gunicorn is much
    more robust than the development server that comes bundled with Django.
    """

    if collectstatic:
        django_manage('collectstatic')
    ctx.run('gunicorn {project_root}.wsgi -b {bind} '
            '--workers {workers} '
            '--name {project}-server'.format(
        project_root=project_root(),
        bind=bind,
        workers=workers,
        project=get_option('options', 'pyname').replace('_', '-')
    ))

def deploy_docs():
    """
    Based on https://gist.github.com/domenic/ec8b0fc8ab45f39403dd
    """
    run('rm -rf ./site/')
    build_docs()
    with util.cd('./site/'):
        run('git init')
        run('echo ".*pyc" > .gitignore')
        run('git config user.name "Travis CI"')
        run('git config user.email "%s"' % os.environ['EMAIL'])
        run('git add .')
        run('git commit -m "Deploy to GitHub Pages"')
        run(
            'git push --force --quiet "https://{GH_TOKEN}@{GH_REF}" '
            'master:gh-pages > /dev/null 2>&1'.format(
                GH_TOKEN=os.environ['GH_TOKEN'],
                GH_REF=os.environ['GH_REF'],
            )
        )

def lint():
    run("flake8 django_make_app tests")

def test():
    run("py.test --verbose --showlocals tests/")

def test_all():
    run("tox")

def test_cov():
    run("py.test --verbose --showlocals --cov=django_make_app tests/")

def test_setuptools():
    run("python setup.py test")

def coverage():
    run("coverage run --source django_make_app setup.py test")
    run("coverage report -m")
    run("coverage html")

def install_requirements():
    run("pip install -r requirements.txt --upgrade --use-wheel")

def test_install():
    run("pip uninstall django_make_app --yes", warn=True)

    run("pip install --use-wheel --no-index --find-links dist django_make_app")
    run("pip uninstall django_make_app --yes")

def build():
    run("python setup.py check --verbose --strict --restructuredtext")

    run("python setup.py build")
    run("python setup.py sdist")
    run("python setup.py bdist_wheel")

def publish():
    run('python setup.py sdist upload -r pypi')
    run('python setup.py bdist_wheel upload -r pypi')

def test(script):
    run("python " + script)

def reset():
    run("rm -rf " + plex_home + "/Plug-in\ Support/Caches/com.plexapp.plugins." + plugin_name)
    run("rm -rf " + plex_home + "/Plug-in\ Support/Data/com.plexapp.plugins." + plugin_name)
    run("rm -rf " + plex_home + "/Plug-in\ Support/Preferences/com.plexapp.plugins." + plugin_name + ".xml")
    # run("rm -rf " + plugin_dir)

    print("Plugin was reset.")

def copy(plugin_dir):
    run("mkdir -p " + plugin_dir + "/Contents/Code")
    run("mkdir -p " + plugin_dir + "/Contents/Libraries/Shared")

    run("cp -R Contents/* " + plugin_dir + "/Contents")
    run("cp -R Contents/Libraries/Shared/* " + plugin_dir + "/Contents/Libraries/Shared")

    print("Files were copied.")

def reload():
    import urllib2

    with open(os.getenv("HOME") + "/Dropbox/.plex-token") as f:
        content = f.readlines()[0].strip()

    url = "http://127.0.0.1:32400/:/plugins/com.plexapp.system/restart?X-Plex-Token=" + content
    urllib2.urlopen(url).read()

    print("Server was restarted.")

    run("tail -f ~/Library/Logs/Plex\ Media\ Server/PMS\ Plugin\ Logs/com.plexapp.plugins." + plugin_name + ".log")

def zip(archive):
    run("cd build && zip -r " + archive + " .")

def clean():
    run("rm -rf build")

def build():
    clean()

    run("mkdir -p build/" + bundle_name)

    copy('build/' + bundle_name)
    zip(archive)

def plex_uninstall():
    run("rm -rf ~/Library/Application Support/Plex Media Server/")
    run("rm -rf ~/Library/Caches/PlexMediaServer/")
    run("rm -rf ~/Library/Preferences/com.plexapp.plexmediaserver.plist")

def prepare(ctx):
    invoke.run('ansible-playbook playbook_prepare.yml')

def deploy_rtr00(ctx):
    invoke.run('ansible-playbook playbook_configure.yml --limit rtr00')

def deploy_rtr01(ctx):
    invoke.run('ansible-playbook playbook_configure.yml --limit rtr01')

def verify(ctx):
    invoke.run('ansible-playbook playbook_facts.yml')

def get_requests(ctx):
    results = invoke.run('../../pinder/manage.py cli --all')

def clean(ctx):
    """Remove created files and caches."""
    run("rm -rf CloudFormation.json")
    run("find . \( -name '__pycache__' -o -name '*.pyc' \) -print -delete")

def build_docs():
    generate_api_reference()
    run('mkdocs build')

def serve_docs():
    generate_api_reference()

    target_cmd = (
        'watchmedo shell-command -R -c '
        '"invoke generate-api-reference" pydeform docs'
    )
    p = threading.Thread(target=run, args=(target_cmd,))
    p.daemon = True
    p.start()

    run('mkdocs serve')

def migrate(ctx, make=False, app=None, runserver=False):
    """
    Run manage.py makemigrations/migrate commands.
    """

    app_suffix = ' %s' % app if app else ''
    if make or io.yn_input('Execute makemigrations? ') == 'yes':
        run("python src/manage.py makemigrations" + app_suffix)
    run("python src/manage.py migrate" + app_suffix)
    if runserver or io.yn_input('Execute runserver? ') == 'yes':
        run('python src/manage.py runserver')

def run(ctx, bind='localhost', port=8000):
    """
    Starts the development server for local testing.
    """

    tail = ''
    if bind and port:
        tail = '%s:%s' % (bind, port)
    elif bind:
        tail = str(bind)
    elif port:
        tail = str(port)
    django_manage('runserver' + tail)

def project_root():
    """
    Return the project root.
    """

    try:
        return get_option('django', 'project_root')
    except ValueError:
        pass

    # No configuration was found. We need to search for a django project in the
    # package tree. The first option is that the project adopts the
    # python-boilerplate layout for a django project:
    pyname = get_option('options', 'pyname')
    if os.path.exists(os.path.join('src', pyname, 'site')):
        return pyname + '.site'

    # Secondly, it might be a layout for a django app. In this case, the
    # project root points to the test project inside the "tests" folder
    if os.path.exists(os.path.join('src', pyname, 'tests', 'site')):
        return pyname + '.tests.site'

    # Everything fails! Quit!
    raise RuntimeError(
        'Could not determine the project root. Is this a really django '
        'project? Please run `python-boilerplate django-project` or '
        '`python-boilerplate django-app` in order to create a project '
        'structure.'
    )

def build(ctx, no_docs=False):
    """
    Build python package and docs.
    """

    util.add_src_to_python_path()
    run("python setup.py build")
    if not no_docs:
        run("python setup.py build_sphinx")

def get_current_schema(config):
    try:
        cqlsh = run_cqlsh(config, command="DESCRIBE " + config['keyspace'])
        out = run(cqlsh, hide='stdout')
    except Exception as e:
        click.secho("Unable to get the current DB schema: {}".format(e), fg='red')
        sys.exit()
    return out.stdout

def verun(cmd):
    """Runs a command inside the virtual environment without
    invoking it

    Args:
        cmd(str): Command to be run in the virtual env
    Returns:
         None
    """
    run('pew in {0} {1}'.format(package_name(), cmd))

def new():
    """Creates a new virtual environment with the package name.
    """
    run('pew new --dont-activate --python={0} '
        '{1}'.format(python_bin, package_name()))
    verun('pip install --upgrade wheel')
    verun('pip install --upgrade pip')

def remove():
    """Removes the virtual environment with the package name.
    """
    run('pew rm {0}'.format(package_name()))

def bash(cmd):
    """Execute command using /bin/bash instead of /bin/sh."""
    return run('/bin/bash -c "{0}"'.format(cmd))

def mkdir(*paths):
    """Create directories and any parent directory that doesn't exist"""
    for path in paths:
        run('mkdir -p {0}'.format(path))

def readout(cmd):
    """A command for getting the package name from
    setup.py file
    Args:
        cmd (str): 'python setup.py --name'
    Returns:
        str: package name
    """
    return run(cmd, hide='out').stdout.strip()

def lint():
    """check style with flake8"""
    run("flake8 {PROJECT_NAME}/ tests/".format(PROJECT_NAME=PROJECT_NAME))

def test():
    run("py.test")

def test_all():
    """run tests on every Python version with tox"""
    run("tox")

def check():
    """Check setup"""
    run("python setup.py --no-user-cfg --verbose check --metadata --restructuredtext --strict")

def coverage():
    """check code coverage quickly with the default Python"""
    run("coverage run --source {PROJECT_NAME} -m py.test".format(PROJECT_NAME=PROJECT_NAME))
    run("coverage report -m")
    run("coverage html")

    if sys.stdout.isatty():
        # Running in a real terminal
        webbrowser.open('file://' + os.path.realpath("htmlcov/index.html"), new=2)

def build():
    """build package"""
    run("python setup.py build")
    run("python setup.py sdist")
    run("python setup.py bdist_wheel")

def publish():
    """publish package"""
    warnings.warn("Deprecated", DeprecationWarning, stacklevel=2)

    check()
    run('python setup.py sdist upload -r pypi')  # Use python setup.py REGISTER
    run('python setup.py bdist_wheel upload -r pypi')

def publish_twine():
    """publish package"""
    check()
    run('twine upload dist/* --skip-existing')

def publish_test():
    """publish package"""
    check()
    run('python setup.py sdist upload -r https://testpypi.python.org/pypi')  # Use python setup.py REGISTER
    run('python setup.py bdist_wheel upload -r https://testpypi.python.org/pypi')