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
python类task()的实例源码
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...')
