Did I find the right examples for you? yes no      Crawl my project      Python Jobs

All Samples(12)  |  Call(10)  |  Derive(0)  |  Import(2)

        def multi_args(p, cmd='celery worker', append='', prefix='', suffix=''):
    names = p.values
    options = dict(p.options)
    passthrough = p.passthrough
    ranges = len(names) == 1
    if ranges:
        try:
            noderange = int(names[0])
        except ValueError:
            pass
        else:
            names = [str(n) for n in range(1, noderange + 1)]
            prefix = 'celery'
    cmd = options.pop('--cmd', cmd)
    append = options.pop('--append', append)
    hostname = options.pop('--hostname',
                           options.pop('-n', socket.gethostname()))
    prefix = options.pop('--prefix', prefix) or ''
    suffix = options.pop('--suffix', suffix) or hostname
    if suffix in ('""', "''"):
        suffix = ''

    for ns_name, ns_opts in list(items(p.namespaces)):
        if ',' in ns_name or (ranges and '-' in ns_name):
            for subns in parse_ns_range(ns_name, ranges):
                p.namespaces[subns].update(ns_opts)
            p.namespaces.pop(ns_name)

    # Numbers in args always refers to the index in the list of names.
    # (e.g. `start foo bar baz -c:1` where 1 is foo, 2 is bar, and so on).
    for ns_name, ns_opts in list(items(p.namespaces)):
        if ns_name.isdigit():
            ns_index = int(ns_name) - 1
            if ns_index < 0:
                raise KeyError('Indexes start at 1 got: %r' % (ns_name, ))
            try:
                p.namespaces[names[ns_index]].update(ns_opts)
            except IndexError:
                raise KeyError('No node at index %r' % (ns_name, ))

    for name in names:
        this_suffix = suffix
        if '@' in name:
            this_name = options['-n'] = name
            nodename, this_suffix = nodesplit(name)
            name = nodename
        else:
            nodename = '%s%s' % (prefix, name)
            this_name = options['-n'] = '%s@%s' % (nodename, this_suffix)
        expand = abbreviations({'%h': this_name,
                                '%n': name,
                                '%N': nodename,
                                '%d': this_suffix})
        argv = ([expand(cmd)] +
                [format_opt(opt, expand(value))
                 for opt, value in items(p.optmerge(name, options))] +
                [passthrough])
        if append:
            argv.append(expand(append))
        yield multi_args_t(this_name, argv, expand, name)
        


src/c/e/celery-HEAD/celery/tests/bin/test_multi.py   celery(Download)
import sys
 
from celery.bin.multi import (
    main,
    MultiTool,
            '--', '.disable_rate_limits=1',
        ])
        it = multi_args(p, cmd='COMMAND', append='*AP*',
                        prefix='*P*', suffix='*S*')
        names = list(it)
        self.assertEqual(expand('%h'), '*P*jerry@*S*')
        self.assertEqual(expand('%n'), '*P*jerry')
        names2 = list(multi_args(p, cmd='COMMAND', append='',
                      prefix='*P*', suffix='*S*'))
        self.assertEqual(names2[0][1][-1], '-- .disable_rate_limits=1')
 
        p2 = NamespacedOptionParser(['10', '-c:1', '5'])
        names3 = list(multi_args(p2, cmd='COMMAND'))
            )
 
        names4 = list(multi_args(p2, cmd='COMMAND', suffix='""'))
        self.assertEqual(len(names4), 10)
        self.assertEqual(

src/c/e/celery-3.1.11/celery/tests/bin/test_multi.py   celery(Download)
import sys
 
from celery.bin.multi import (
    main,
    MultiTool,
            '--', '.disable_rate_limits=1',
        ])
        it = multi_args(p, cmd='COMMAND', append='*AP*',
                        prefix='*P*', suffix='*S*')
        names = list(it)
        self.assertEqual(expand('%h'), '*P*jerry@*S*')
        self.assertEqual(expand('%n'), 'jerry')
        names2 = list(multi_args(p, cmd='COMMAND', append='',
                      prefix='*P*', suffix='*S*'))
        self.assertEqual(names2[0][1][-1], '-- .disable_rate_limits=1')
 
        gethostname.return_value = 'example.com'
        p2 = NamespacedOptionParser(['10', '-c:1', '5'])
        names3 = list(multi_args(p2, cmd='COMMAND'))
            )
 
        names4 = list(multi_args(p2, cmd='COMMAND', suffix='""'))
        self.assertEqual(len(names4), 10)
        self.assertEqual(