sontek / pyramid_celery / 87

pending completion

Build # 87

Build Type

Pull #83

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/sontek/pyramid_celery/commit/<a class=hub.com/sontek/pyramid_celery/commit/3e07a27be96bceef40e322e9c68f6153aff2fdf4">3e07a27be<a href="https://github.com/sontek/pyramid_celery/commit/3e07a27be96bceef40e322e9c68f6153aff2fdf4">">Merge </a><a class="double-link" href="https://github.com/sontek/pyramid_celery/commit/<a class="double-link" href="https://github.com/sontek/pyramid_celery/commit/80d0060f47b1dd51682ba99a1fe58aeb7471bc40">80d0060f4</a>">80d0060f4</a><a href="https://github.com/sontek/pyramid_celery/commit/3e07a27be96bceef40e322e9c68f6153aff2fdf4"> into 75064bec0">75064bec0</a>

Pull Request Pull Request #83: this modernizes pyramid_celery

Run Details

9 of 9 new or added lines in 2 files covered. (100.0%)

143 of 152 relevant lines covered (94.08%)

4.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.5

/pyramid_celery/__init__.py

from celery import Celery
from celery import signals
from celery import VERSION as celery_version

try:
    from celery.bin.base import Option
except ImportError:
    pass

from pyramid.paster import bootstrap, setup_logging
from pyramid_celery.loaders import INILoader
from pyramid.settings import asbool


def add_preload_arguments(parser):
    parser.add_argument(
        '-i', '--ini', default=None,
        help='Paste ini configuration file.'
    )
    parser.add_argument(
        '--ini-var', default=None,
        help='Comma separated list of key=value to pass to ini.'
    )


celery_app = Celery()
if celery_version.major > 3:
    celery_app.user_options['preload'].add(add_preload_arguments)
else:
    celery_app.user_options['preload'].add(Option(
        '-i', '--ini', default=None,
        help='Paste ini configuration file.'
    ))
    celery_app.user_options['preload'].add(Option(
        '--ini-var', default=None,
        help='Comma separated list of key=value to pass to ini.'
    ))
ini_file = None


def boolify(config, *names):
    """Make config variables boolean.

    Celery wants ``False`` instead of ``"false"`` for CELERY_ALWAYS_EAGER.
    """

    for n in names:
        if n in config:
            config[n] = asbool(config[n])


def configure_logging(*args, **kwargs):
    setup_logging(ini_file)


def setup_app(app, root, request, registry, closer, ini_location):
    loader = INILoader(celery_app, ini_file=ini_location)
    celery_config = loader.read_configuration()
    #: TODO: There might be other variables requiring special handling
    boolify(
        celery_config,
        'task_always_eager',
        'enable_utc',
        'result_persistent',
    )

    if asbool(celery_config.get('use_celeryconfig', False)) is True:
        config_path = 'celeryconfig'
        celery_app.config_from_object(config_path)
    else:
        hijack_key = 'worker_hijack_root_logger'

        # TODO: Couldn't find a way with celery to do this
        hijack_logger = asbool(
            celery_config.get(hijack_key, False)
        )

        celery_config[hijack_key] = hijack_logger

        if hijack_logger is False:
            global ini_file
            ini_file = ini_location
            signals.setup_logging.connect(configure_logging)

        celery_app.config_from_object(celery_config)

    celery_app.conf.update({'PYRAMID_APP': app})
    celery_app.conf.update({'PYRAMID_ROOT': root})
    celery_app.conf.update({'PYRAMID_REQUEST': request})
    celery_app.conf.update({'PYRAMID_REGISTRY': registry})
    celery_app.conf.update({'PYRAMID_CLOSER': closer})


@signals.user_preload_options.connect
def on_preload_parsed(options, **kwargs):
    ini_location = options['ini']
    ini_vars = options['ini_var']

    if ini_location is None:
        print('You must provide the paste --ini argument')
        exit(-1)

    options = {}
    try:
        if ini_vars is not None:
            for pairs in ini_vars.split(','):
                key, value = pairs.split('=')
                options[key] = value

            env = bootstrap(ini_location, options=options)
        else:
            env = bootstrap(ini_location)
    except:
        import traceback
        traceback.print_exc()
        exit(-1)

    registry = env['registry']
    app = env['app']
    root = env['root']
    request = env['request']
    closer = env['closer']
    setup_app(app, root, request, registry, closer, ini_location)


def configure(config, ini_location):
    setup_app(
        None,
        None,
        None,
        config.registry,
        None,
        ini_location
    )


def includeme(config):
    config.add_directive('configure_celery', configure)

1	from celery import Celery	5✔
2	from celery import signals	5✔
3	from celery import VERSION as celery_version	5✔
4
5	try:	5✔
6	from celery.bin.base import Option	5✔
7	except ImportError:	5✔
8	pass	5✔
9
10	from pyramid.paster import bootstrap, setup_logging	5✔
11	from pyramid_celery.loaders import INILoader	5✔
12	from pyramid.settings import asbool	5✔
13
14
15	def add_preload_arguments(parser):	5✔
16	parser.add_argument(	×
17	'-i', '--ini', default=None,
18	help='Paste ini configuration file.'
19	)
20	parser.add_argument(	×
21	'--ini-var', default=None,
22	help='Comma separated list of key=value to pass to ini.'
23	)
24
25
26	celery_app = Celery()	5✔
27	if celery_version.major > 3:	5✔
28	celery_app.user_options['preload'].add(add_preload_arguments)	5✔
29	else:
30	celery_app.user_options['preload'].add(Option(	×
31	'-i', '--ini', default=None,
32	help='Paste ini configuration file.'
33	))
34	celery_app.user_options['preload'].add(Option(	×
35	'--ini-var', default=None,
36	help='Comma separated list of key=value to pass to ini.'
37	))
38	ini_file = None	5✔
39
40
41	def boolify(config, *names):	5✔
42	"""Make config variables boolean.
43
44	Celery wants ``False`` instead of ``"false"`` for CELERY_ALWAYS_EAGER.
45	"""
46
47	for n in names:	5✔
48	if n in config:	5✔
49	config[n] = asbool(config[n])	×
50
51
52	def configure_logging(args, *kwargs):	5✔
53	setup_logging(ini_file)	5✔
54
55
56	def setup_app(app, root, request, registry, closer, ini_location):	5✔
57	loader = INILoader(celery_app, ini_file=ini_location)	5✔
58	celery_config = loader.read_configuration()	5✔
59	#: TODO: There might be other variables requiring special handling
60	boolify(	5✔
61	celery_config,
62	'task_always_eager',
63	'enable_utc',
64	'result_persistent',
65	)
66
67	if asbool(celery_config.get('use_celeryconfig', False)) is True:	5✔
68	config_path = 'celeryconfig'	5✔
69	celery_app.config_from_object(config_path)	5✔
70	else:
71	hijack_key = 'worker_hijack_root_logger'	5✔
72
73	# TODO: Couldn't find a way with celery to do this
74	hijack_logger = asbool(	5✔
75	celery_config.get(hijack_key, False)
76	)
77
78	celery_config[hijack_key] = hijack_logger	5✔
79
80	if hijack_logger is False:	5✔
81	global ini_file
82	ini_file = ini_location	5✔
83	signals.setup_logging.connect(configure_logging)	5✔
84
85	celery_app.config_from_object(celery_config)	5✔
86
87	celery_app.conf.update({'PYRAMID_APP': app})	5✔
88	celery_app.conf.update({'PYRAMID_ROOT': root})	5✔
89	celery_app.conf.update({'PYRAMID_REQUEST': request})	5✔
90	celery_app.conf.update({'PYRAMID_REGISTRY': registry})	5✔
91	celery_app.conf.update({'PYRAMID_CLOSER': closer})	5✔
92
93
94	@signals.user_preload_options.connect	5✔
95	def on_preload_parsed(options, **kwargs):	2✔
96	ini_location = options['ini']	5✔
97	ini_vars = options['ini_var']	5✔
98
99	if ini_location is None:	5✔
100	print('You must provide the paste --ini argument')	5✔
101	exit(-1)	5✔
102
103	options = {}	5✔
104	try:	5✔
105	if ini_vars is not None:	5✔
106	for pairs in ini_vars.split(','):	5✔
107	key, value = pairs.split('=')	5✔
108	options[key] = value	5✔
109
110	env = bootstrap(ini_location, options=options)	5✔
111	else:
112	env = bootstrap(ini_location)	5✔
113	except:	×
114	import traceback	×
115	traceback.print_exc()	×
116	exit(-1)	×
117
118	registry = env['registry']	5✔
119	app = env['app']	5✔
120	root = env['root']	5✔
121	request = env['request']	5✔
122	closer = env['closer']	5✔
123	setup_app(app, root, request, registry, closer, ini_location)	5✔
124
125
126	def configure(config, ini_location):	5✔
127	setup_app(	5✔
128	None,
129	None,
130	None,
131	config.registry,
132	None,
133	ini_location
134	)
135
136
137	def includeme(config):	5✔
138	config.add_directive('configure_celery', configure)	5✔

sontek / pyramid_celery / 87

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous