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

All Samples(20)  |  Call(20)  |  Derive(0)  |  Import(0)
Expands the :setting:`CELERY_ROUTES` setting.

        def prepare(routes):
    """Expands the :setting:`CELERY_ROUTES` setting."""

    def expand_route(route):
        if isinstance(route, dict):
            return MapRoute(route)
        if isinstance(route, string_t):
            return mlazy(instantiate, route)
        return route

    if routes is None:
        return ()
    if not isinstance(routes, (list, tuple)):
        routes = (routes, )
    return [expand_route(route) for route in routes]
        


src/a/n/antisocial-HEAD/ve/lib/python2.7/site-packages/celery/tests/app/test_routes.py   antisocial(Download)
    def test_lookup_takes_first(self):
        R = routes.prepare(({mytask.name: {'queue': 'bar'}},
                            {mytask.name: {'queue': 'foo'}}))
        router = Router(R, current_app.amqp.queues)
        self.assertEqual(router.route({}, mytask.name,
                         args=[1, 2], kwargs={})['queue'].name, 'bar')
 
    @with_queues()
    def test_expands_queue_in_options(self):
        R = routes.prepare(())
    def test_lookup_paths_traversed(self):
        R = routes.prepare(({'celery.xaza': {'queue': 'bar'}},
                            {mytask.name: {'queue': 'foo'}}))
        router = Router(R, current_app.amqp.queues)
        self.assertEqual(router.route({}, mytask.name,
        R = [{'foo': 'bar'},
             'celery.datastructures.LRUCache', o]
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)
        self.assertIsInstance(maybe_promise(p[1]), LRUCache)
    def test_prepare_item_is_dict(self):
        R = {'foo': 'bar'}
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)
 

src/f/i/firefox-flicks-HEAD/vendor-local/lib/python/celery/tests/app/test_routes.py   firefox-flicks(Download)
    def test_lookup_takes_first(self):
        R = routes.prepare(({mytask.name: {'queue': 'bar'}},
                            {mytask.name: {'queue': 'foo'}}))
        router = Router(R, current_app.amqp.queues)
        self.assertEqual(router.route({}, mytask.name,
                         args=[1, 2], kwargs={})['queue'].name, 'bar')
 
    @with_queues()
    def test_expands_queue_in_options(self):
        R = routes.prepare(())
    def test_lookup_paths_traversed(self):
        R = routes.prepare(({'celery.xaza': {'queue': 'bar'}},
                            {mytask.name: {'queue': 'foo'}}))
        router = Router(R, current_app.amqp.queues)
        self.assertEqual(router.route({}, mytask.name,
        R = [{'foo': 'bar'},
             'celery.datastructures.LRUCache', o]
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)
        self.assertIsInstance(maybe_promise(p[1]), LRUCache)
    def test_prepare_item_is_dict(self):
        R = {'foo': 'bar'}
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)
 

src/c/e/celery-3.1.11/celery/tests/app/test_routes.py   celery(Download)
    def test_lookup_takes_first(self):
        set_queues(self.app, foo=self.a_queue, bar=self.b_queue)
        R = routes.prepare(({self.mytask.name: {'queue': 'bar'}},
                            {self.mytask.name: {'queue': 'foo'}}))
        router = Router(self.app, R, self.app.amqp.queues)
    def test_expands_queue_in_options(self):
        set_queues(self.app)
        R = routes.prepare(())
        router = Router(
            self.app, R, self.app.amqp.queues, create_missing=True,
            **{self.app.conf.CELERY_DEFAULT_QUEUE: self.d_queue}
        )
        R = routes.prepare((
            {'celery.xaza': {'queue': 'bar'}},
            {self.mytask.name: {'queue': 'foo'}}
    def test_prepare(self):
        o = object()
        R = [{'foo': 'bar'},
             'celery.utils.functional.LRUCache', o]
        p = routes.prepare(R)
    def test_prepare_item_is_dict(self):
        R = {'foo': 'bar'}
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)
 

src/c/e/celery-HEAD/celery/tests/app/test_routes.py   celery(Download)
    def test_lookup_takes_first(self):
        set_queues(self.app, foo=self.a_queue, bar=self.b_queue)
        R = routes.prepare(({self.mytask.name: {'queue': 'bar'}},
                            {self.mytask.name: {'queue': 'foo'}}))
        router = Router(self.app, R, self.app.amqp.queues)
    def test_expands_queue_in_options(self):
        set_queues(self.app)
        R = routes.prepare(())
        router = Router(
            self.app, R, self.app.amqp.queues, create_missing=True,
            **{self.app.conf.CELERY_DEFAULT_QUEUE: self.d_queue}
        )
        R = routes.prepare((
            {'celery.xaza': {'queue': 'bar'}},
            {self.mytask.name: {'queue': 'foo'}}
    def test_prepare(self):
        o = object()
        R = [{'foo': 'bar'},
             'celery.utils.functional.LRUCache', o]
        p = routes.prepare(R)
    def test_prepare_item_is_dict(self):
        R = {'foo': 'bar'}
        p = routes.prepare(R)
        self.assertIsInstance(p[0], routes.MapRoute)