Celery用户指引--------Application

>>> from celery import Celery
>>> app = Celery()
>>> app
<Celery __main__:0x100469fd0>

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.name
__main__.add

>>> app.tasks['__main__.add']
<@task: __main__.add>

from celery import Celery
app = Celery()

@app.task
def add(x, y): return x + y

if __name__ == '__main__':
app.worker_main()

>>> from tasks import add
>>> add.name
tasks.add

>>> app = Celery('tasks')
>>> app.main
'tasks'

>>> @app.task
... def add(x, y):
...     return x + y

>>> add.name
tasks.add

>>> app.conf.timezone
'Europe/London'

>>> app.conf.enable_utc = True

>>> app.conf.update(
...     enable_utc=True,
...     timezone='Europe/London',
...)

app.add_defaults()

app.config_from_object()方法从一个配置对象中加载配置

配置对象可以是一个模块或者任何含有配置属性的对象。

注意，任何先前设置的配置在调用config_from_object后都会被重新设置。如果你想添加额外的配置，你需要在调用这个方法之后。

举例1：使用模块和名称

app.config_from_object方法可以使用合法的python模块名，甚至还可以使用一个python属性，比如："celeryconfig","myproj.config.celery"或者"myproj.config:CeleryConfig":

from celery import Celeryapp = Celery()app.config_from_object('celeryconfig')

celeryconfig.py

enable_utc = Truetimezone = 'Europe/London'

举例2：传递一个实际的模块对象

你也可以传递一个已经import导入的模块对象，但是这种做法是不推荐使用的。

说明：

推荐使用模块的名称，因为这意味着当使用prefork池时这个模块不必被序列化。如果你遇到了配置问题或者pickle错误，请使用模块名称代替使用导入的模块。

import celeryconfigfrom celery import Celeryapp = Celery()app.config_from_object(celeryconfig)

from celery import Celeryapp = Celery()class Config:enable_utc = Truetimezone = 'Europe/London'app.config_from_object(Config)# or using the fully qualified name of the object:#   app.config_from_object('module:Config')

app.config_from_envvar()从环境变量中获取配置模块的名称。

比如---从环境变量“CELERY_CONFIG_MODULE”指定的模块名中加载配置。

import osfrom celery import Celery#: Set default configuration module nameos.environ.setdefault('CELERY_CONFIG_MODULE', 'celeryconfig')app = Celery()app.config_from_envvar('CELERY_CONFIG_MODULE')

$ CELERY_CONFIG_MODULE="celeryconfig.prod" celery worker -l info

humanize()

>>> app.conf.humanize(with_defaults=False, censored=True)

table()

>>> app.conf.table(with_defaults=False, censored=True)

API

TOKEN

KEY

SECRET

PASS

SIGNATURE

DATABASE

>>> @app.task>>> def add(x, y):...    return x + y>>> type(add)<class 'celery.local.PromiseProxy'>>>> add.__evaluated__()False>>> add        # <-- causes repr(add) to happen<@task: __main__.add>>>> add.__evaluated__()True

from celery.task import Task   # << OLD Task base class.from celery import Task        # << NEW base class.

from celery import current_appclass Scheduler(object):def run(self):app = current_app

class Scheduler(object):def __init__(self, app):self.app = app

from celery.app import app_or_defaultclass Scheduler(object):def __init__(self, app=None):self.app = app_or_default(app)

$ CELERY_TRACE_APP=1 celery worker -l info

def hello(to):return 'hello {0}'.format(to)>>> from celery.execute import apply_async>>> apply_async(hello, ('world!',))

from celery.task import Taskfrom celery.registry import tasksclass Hello(Task):send_error_emails = Truedef run(self, to):return 'hello {0}'.format(to)tasks.register(Hello)>>> Hello.delay('world!')

from celery.task import task@task(send_error_emails=True)def hello(x):return 'hello {0}'.format(to)

@app.task(base=OtherTask):def add(x, y):return x + y

from celery import Taskclass DebugTask(Task):def __call__(self, *args, **kwargs):print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))return super(DebugTask, self).__call__(*args, **kwargs)

@app.task(base=DebugTask)def add(x, y):return x + y

>>> from celery import Celery, Task>>> app = Celery()>>> class MyBaseTask(Task):...    send_error_emails = True>>> app.Task = MyBaseTask>>> app.Task<unbound MyBaseTask>>>> @app.task... def add(x, y):...     return x + y>>> add<@task: __main__.add>>>> add.__class__.mro()[<class add of <Celery __main__:0x1012b4410>>,<unbound MyBaseTask>,<unbound Task>,<type 'object'>]