本文整理汇总了Python中workflow.background.run_in_background函数的典型用法代码示例。如果您正苦于以下问题:Python run_in_background函数的具体用法?Python run_in_background怎么用?Python run_in_background使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了run_in_background函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: run
def run(self, action, env=False):
action = Action(action)
if not filter(lambda x: x.name == action.name, self.actions):
raise Exception('Action {0} not found. Instance changed state?'
.format(action))
task_name = 'exec{0}'.format(hash(self.vagrantfile_path))
if action.name in ('rdp', 'ssh'):
run_vagrant('{action} {machine_id}'
.format(action=action.name, machine_id=self.key))
return
cur_dir = os.path.dirname(os.path.realpath(__file__))
exec_path = os.path.join(cur_dir, 'execute.py')
cmd = ['python', exec_path, '--action', action.name]
if action.flags:
cmd += ['-f'] + action.flags
if not env:
cmd += ['--name', self.name]
logger.debug('Running in background: %s', cmd)
new_env = os.environ.copy()
new_env['HOME'] = os.path.expanduser('~')
new_env['PATH'] = Workflow().settings['PATH']['VAR']
run_in_background(task_name,
cmd,
env=new_env,
cwd=self.vagrantfile_path)
开发者ID:m1keil,项目名称:alfred-vagrant-workflow,代码行数:30,代码来源:vagrant.py
示例2: main
def main(wf):
import os
# Argument parsing configuration
parser = argparse.ArgumentParser()
parser.add_argument('--setlang', dest='lang', nargs='?', default=0)
parser.add_argument('query', nargs='?', default=None)
args = parser.parse_args(wf.args)
if args.lang == '1':
global lang
lang = 1 # Set global lang
# Call different functions filtering by argument
if 'showmanga:' in args.query: # Search and show list of results
mid = args.query.split('showmanga:')[1] # Parse the unique manga id
search_by_mid(wf, mid)
elif 'downloadmanga:' in args.query: # Perform a massive download of selected manga
mid = args.query.split('downloadmanga:')[1]
# TODO check if there is at least 1 element
notify.notify("Background download started!",
"Multithreading download started. You will be notified when completed.", sound='Submarine')
wf.add_item("Download started", "At the end you will be notified in the notification center.",
valid=False, arg="", autocomplete="") # Autocomplete will restore the clear command
# Run massive download in background (~ subprocess.call)
run_in_background('download', ['/usr/bin/python', wf.workflowfile('mangaedenAPI.py'), 'dmanga:' + mid])
wf.send_feedback()
elif 'dmanga:' in args.query:
# TODO better error handling while background download is running
mid = args.query.split('dmanga:')[1]
download_manga(mid)
else:
# Search on mangaeden json list
query = args.query.lower()
search_by_query(wf, query)
开发者ID:Emmunaf,项目名称:MangaEden-Search,代码行数:33,代码来源:mangaedenAPI.py
示例3: main
def main(wf):
"""Run workflow Script Filter.
Args:
wf (workflow.Workflow): Current Workflow object.
"""
global ureg
ureg = UnitRegistry(wf.decode(DEFAULT_UNIT_DEFINITIONS))
ureg.default_format = 'P'
wf.magic_arguments['appkey'] = open_currency_instructions
if not len(wf.args):
return
query = wf.args[0] # .lower()
log.debug('query : %s', query)
handle_update(wf)
# Create data files if necessary
bootstrap(wf)
# Add workflow and user units to unit registry
register_units()
# Notify of available update
if wf.update_available:
wf.add_item('A newer version is available',
'Action this item to download & install the new version',
autocomplete='workflow:update',
icon=ICON_UPDATE)
# Load cached data
exchange_rates = wf.cached_data(CURRENCY_CACHE_NAME, max_age=0)
if exchange_rates: # Add exchange rates to conversion database
register_exchange_rates(exchange_rates)
if not wf.cached_data_fresh(CURRENCY_CACHE_NAME, CURRENCY_CACHE_AGE):
# Update currency rates
cmd = ['/usr/bin/python', wf.workflowfile('currency.py')]
run_in_background('update', cmd)
wf.rerun = 0.5
if is_running('update'):
wf.rerun = 0.5
if exchange_rates is None: # No data cached yet
wf.add_item(u'Fetching exchange rates…',
'Currency conversions will be momentarily possible',
icon=ICON_INFO)
else:
wf.add_item(u'Updating exchange rates…',
icon=ICON_INFO)
return convert(query)
开发者ID:kjunggithub,项目名称:dotfiles,代码行数:56,代码来源:convert.py
示例4: generate_all_icons
def generate_all_icons():
"""Callback for magic argument"""
if background.is_running('icongen'):
return 'Generation already in progress.'
background.run_in_background(
'icongen',
['/usr/bin/python', wf.workflowfile('icons.py')]
)
return 'Starting icon generation. This may take up to 15 minutes.'
开发者ID:deanishe,项目名称:alfred-unicode,代码行数:10,代码来源:characters.py
示例5: test_run_in_background
def test_run_in_background(self):
"""Run in background"""
cmd = ['sleep', '1']
run_in_background('test', cmd)
sleep(0.5)
self.assertTrue(is_running('test'))
self.assertTrue(os.path.exists(self._pidfile('test')))
self.assertEqual(run_in_background('test', cmd), None)
sleep(0.6)
self.assertFalse(is_running('test'))
self.assertFalse(os.path.exists(self._pidfile('test')))
开发者ID:JT5D,项目名称:Alfred-Popclip-Sublime,代码行数:11,代码来源:test_background.py
示例6: test_run_in_background
def test_run_in_background(self):
"""Run in background"""
cmd = ['sleep', '1']
assert run_in_background('test', cmd) == 0
assert is_running('test')
assert os.path.exists(_pidfile('test'))
# Already running
assert run_in_background('test', cmd) is None
sleep(1.1) # wait for job to finish
assert not is_running('test')
assert not os.path.exists(_pidfile('test'))
开发者ID:deanishe,项目名称:alfred-workflow,代码行数:11,代码来源:test_background.py
示例7: update
def update(self, force=False):
"""Load contacts from cache and update cached data if old."""
# Load cached contacts
self.contacts = wf.cached_data('contacts', max_age=0)
if self.contacts is None:
self.contacts = {}
# Update if required
if not wf.cached_data_fresh('contacts', MAX_CACHE_AGE) or force:
log.debug('Updating contacts cache ...')
cmd = ['/usr/bin/python', wf.workflowfile('update_contacts.py')]
run_in_background('update-contacts', cmd)
开发者ID:deanishe,项目名称:alfred-mailto,代码行数:12,代码来源:contacts.py
示例8: background_sync
def background_sync():
from workflow.background import run_in_background
task_id = 'sync'
# Only runs if another sync is not already in progress
run_in_background(task_id, [
'/usr/bin/env',
'python',
workflow().workflowfile('alfred-wunderlist-workflow.py'),
'pref sync background',
'--commit'
])
开发者ID:amatiasq,项目名称:dotfiles,代码行数:12,代码来源:sync.py
示例9: backgroundSync
def backgroundSync():
from workflow.background import run_in_background
from wunderlist.util import workflow
# Only runs if another sync is not already in progress
run_in_background('sync', [
'/usr/bin/env',
'python',
workflow().workflowfile('alfred-wunderlist-workflow.py'),
':pref sync',
'--commit'
])
开发者ID:Alfr0475,项目名称:alfred.d,代码行数:12,代码来源:sync.py
示例10: update_data
def update_data(update_method):
"""Update project data from 10.000ft"""
log.debug('Starting update')
cmd = ['/usr/bin/python', wf.workflowfile('update.py')]
if update_method == 'force':
cmd.append('--update')
cmd.append('force')
# Update projects data
log.debug('Run update command : {}'.format(cmd))
run_in_background('update', cmd)
return 0
开发者ID:jceelen,项目名称:alfred-10000ft-scripts,代码行数:13,代码来源:10000ft.py
示例11: do_import_search
def do_import_search(wf, url):
"""Parse URL for OpenSearch config."""
ctx = Context(wf)
# ICON_IMPORT = ctx.icon('import')
ICONS_PROGRESS = [
ctx.icon('progress-1'),
ctx.icon('progress-2'),
ctx.icon('progress-3'),
ctx.icon('progress-4'),
]
data = wf.cached_data('import', None, max_age=0, session=True)
if data:
error = data['error']
search = data['search']
# Clear cache data
wf.cache_data('import', None, session=True)
wf.cache_data('import-status', None, session=True)
if error:
wf.add_item(error, icon=ICON_ERROR)
wf.send_feedback()
return
it = wf.add_item(u'Add "{}"'.format(search['name']),
u'↩ to add search',
valid=True,
icon=search['icon'])
for k, v in search.items():
it.setvar(k, v)
else:
progress = int(os.getenv('progress') or '0')
i = progress % len(ICONS_PROGRESS)
picon = ICONS_PROGRESS[i]
log.debug('progress=%d, i=%d, picon=%s', progress, i, picon)
wf.setvar('progress', progress + 1)
if not is_running('import'):
run_in_background('import', ['./searchio', 'fetch', url])
status = wf.cached_data('import-status', None, max_age=0, session=True)
title = status or u'Fetching OpenSearch Configuration …'
wf.rerun = 0.2
wf.add_item(title,
u'Results will be shown momentarily',
icon=picon)
wf.send_feedback()
开发者ID:deanishe,项目名称:alfred-searchio,代码行数:50,代码来源:web.py
示例12: main
def main(wf):
args = Args(wf.args)
actions = wf.cached_data('actions', None, max_age=0)
if wf.update_available:
# Add a notification to top of Script Filter results
wf.add_item(u'New version available',
u'Action this item to install the update',
autocomplete='workflow:update',
icon=ICON_INFO)
if not wf.cached_data_fresh('actions', max_age=CACHE_MAX_AGE):
cmd = ['/usr/bin/python', wf.workflowfile('alfredhelp.py'), '--scan']
run_in_background(u'scan', cmd)
if is_running(u'scan'):
wf.add_item(
title=u'Scanning alfred workflows...',
valid=False,
icon=ICON_INFO
)
if args.show_keywords and actions:
if args.query:
actions = wf.filter(args.query, actions, key=search_key)
for action in actions:
argument = action.keyword
if action.add_space:
argument += u' '
wf.add_item(
title=u'{keyword} - {title}'.format(keyword=action.keyword, title=action.title),
subtitle=action.subtitle,
icon=action.icon,
arg=argument,
valid=True
)
elif args.scan:
def get_posts():
return scan(path.join(wf.alfred_env['preferences'], 'workflows'))
wf.cached_data('actions', get_posts, max_age=CACHE_MAX_AGE)
scan(path.join(wf.alfred_env['preferences'], 'workflows'))
wf.send_feedback()
return 0
开发者ID:qoomon,项目名称:alfred-workflows,代码行数:48,代码来源:alfredhelp.py
示例13: main
def main(wf):
parser = argparse.ArgumentParser()
parser.add_argument('--browse', dest = 'browse', nargs = '?', default = None)
parser.add_argument('--iterm', dest = 'iterm', nargs = argparse.REMAINDER, default = None)
args = parser.parse_args(wf.args)
if args.browse:
out = ['open', args.browse]
run_in_background('browse', out)
return 0
if args.iterm:
out = ['osascript', 'open_iterm.scpt', ' '.join(args.iterm)]
run_in_background('iterm', out)
return 0
开发者ID:pburt,项目名称:alfred-stash,代码行数:16,代码来源:stash_output.py
示例14: main
def main(workflow):
parser = argparse.ArgumentParser()
parser.add_argument("--set-token", dest="api_token", nargs="?", default=None)
parser.add_argument("query", nargs="?", default=None)
arguments = parser.parse_args(workflow.args)
if arguments.api_token:
workflow.save_password("hipchat_api_token", arguments.api_token)
return 0
try:
api_token = workflow.get_password("hipchat_api_token")
except PasswordNotFound:
workflow.add_item(
"No API key set.", "Please use hcsettoken to set your Hipchat API token.", valid=False, icon=ICON_WARNING
)
workflow.send_feedback()
return 0
users = workflow.cached_data("users", None, max_age=0)
if not workflow.cached_data_fresh("users", max_age=60): # 60s
cmd = ["/usr/bin/python", workflow.workflowfile("update.py")]
run_in_background("update", cmd)
if is_running("update"):
logger.debug("updating users")
if arguments.query and users:
users = workflow.filter(arguments.query, users, key=search_terms_for_user, min_score=20)
if not users:
workflow.add_item("Whoops, no users found", icon=ICON_WARNING)
workflow.send_feedback()
return 0
for user in users:
status_icon = get_status_icon(user["presence"]["show"] if user["presence"] else None)
workflow.add_item(
user["name"], user["email"], arg=actionTemplate.format(user["mention_name"]), valid=True, icon=status_icon
)
workflow.send_feedback()
logger.debug("returned {} results".format(len(users)))
开发者ID:deekim,项目名称:alfred-hipchat,代码行数:46,代码来源:hipchat.py
示例15: search_vault_for_query
def search_vault_for_query(self, query):
"""
Search the LastPass vault for an optional passed query.
"""
results = self.wf.cached_data("vault_items", None, max_age=0)
# Start updae script if cache is too old or doesn't exist:
age = int(self.wf.settings["general"]["cache_bust"])
if not self.wf.cached_data_fresh("vault_items", age):
cmd = ["/usr/bin/python", self.wf.workflowfile("update.py")]
run_in_background("update", cmd)
# If a query is passed, filter the results:
if query and results:
results = self.wf.filter(query, results, self.search_item_fields, match_on=MATCH_ALL ^ MATCH_ALLCHARS)
self.log.debug("Search results: {}".format(results))
return results
开发者ID:hbcbh1999,项目名称:lp-vault-manager,代码行数:18,代码来源:utilities.py
示例16: test_kill
def test_kill(self):
"""Kill"""
assert kill('test') is False
cmd = ['sleep', '1']
assert run_in_background('test', cmd) == 0
assert is_running('test')
assert kill('test') is True
sleep(0.3) # give process time to exit
assert not is_running('test')
开发者ID:deanishe,项目名称:alfred-workflow,代码行数:9,代码来源:test_background.py
示例17: run
def run(self, wf):
"""Run workflow."""
self.wf = wf
wf.args # check for magic args
self.keyword = self.wf.settings.get('keyword', DEFAULT_KEYWORD)
args = docopt(__doc__)
log.debug(u'args : %r', args)
# Open Help file
if args.get('--helpfile'):
return self.do_open_help_file()
# Perform search
self.query = wf.decode(args.get('<query>') or '')
# List Smart Folders with custom keywords
if args.get('--config'):
return self.do_configure_folders()
# Was a configured folder passed?
folder = wf.decode(args.get('--folder') or '')
# Get list of Smart Folders. Update in background if necessary.
self.folders = self.wf.cached_data('folders', max_age=0)
if self.folders is None:
self.folders = []
# Update folder list if it's old
if not self.wf.cached_data_fresh('folders', CACHE_AGE_FOLDERS):
log.debug('updating list of Smart Folders in background...')
run_in_background('folders',
['/usr/bin/python',
self.wf.workflowfile('cache.py')])
if is_running('folders'):
self.wf.rerun = 0.5
# Has a specific folder been specified?
if folder:
return self.do_search_in_folder(folder)
return self.do_search_folders()
开发者ID:deanishe,项目名称:alfred-smartfolders,代码行数:42,代码来源:smartfolders.py
示例18: main
def main(wf):
parser = argparse.ArgumentParser()
parser.add_argument('--baseurl', dest='baseurl', nargs='?', default=None)
parser.add_argument('--delimiter', dest='delimiter', nargs='?', default=None)
parser.add_argument('--directory', dest='directory', nargs='?', default=None)
parser.add_argument('--username', dest='username', nargs='?', default=None)
parser.add_argument('--password', dest='password', nargs='?', default=None)
parser.add_argument('--avatars', dest='avatars', action='store_true', default=None)
parser.add_argument('--edit', dest='edit', action='store_true', default=None)
args = parser.parse_args(wf.args)
if args.baseurl:
wf.settings['baseurl'] = args.baseurl.rstrip('/')
return 0
if args.delimiter:
wf.settings['delimiter'] = args.delimiter
return 0
if args.directory:
wf.settings['directory'] = args.directory.rstrip('/')
return 0
if args.username:
wf.settings['username'] = args.username
return 0
if args.password:
wf.save_password('stash_password', args.password)
return 0
if args.avatars:
cmd = ['/usr/bin/python', wf.workflowfile('update_avatars.py')]
run_in_background('update-avatars', cmd)
return 0
if args.edit:
out = ['open', wf.settings_path]
run_in_background('edit', out)
return 0
开发者ID:pburt,项目名称:alfred-stash,代码行数:41,代码来源:stash_settings_output.py
示例19: update
def update(self, force=False):
"""Load apps from cache, update if required"""
self.all_email_apps = self.wf.cached_data('all_apps', max_age=0)
self.system_default_app = self.wf.cached_data('system_default_app',
max_age=0)
if self.all_email_apps is None:
self.all_email_apps = []
if self.system_default_app is None:
self.system_default_app = {}
do_update = False
if force:
do_update = True
elif not self.wf.cached_data_fresh('all_apps', MAX_APP_CACHE_AGE):
do_update = True
elif not self.wf.cached_data_fresh('system_default_app',
MAX_APP_CACHE_AGE):
do_update = True
# Update if required
if do_update:
log.debug('Updating application caches ...')
cmd = ['/usr/bin/python', self.wf.workflowfile('update_apps.py')]
run_in_background('update-apps', cmd)
开发者ID:sunliwen,项目名称:alfred-mailto,代码行数:23,代码来源:client.py
示例20: _update
def _update(self, force=False):
"""Update cached data"""
log.debug('Updating workflow lists...')
args = ['/usr/bin/python',
self.wf.workflowfile('update_workflows.py')]
if force:
args.append('--force-update')
log.debug('update command : {}'.format(args))
retcode = run_in_background('update', args)
if retcode:
log.debug('Update failed with code {}'.format(retcode))
print('Update failed')
return 1
print('Updating workflow list…'.encode('utf-8'))
return 0
开发者ID:ccsouthard,项目名称:config,代码行数:15,代码来源:packal.py
注:本文中的workflow.background.run_in_background函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论