在Python中用Celery安排管理后台工作流

    - celery_uncovered/
      - celery_uncovered/__init__.py
      - celery_uncovered/{toyex,tricks,advex}
      - celery_uncovered/celery.py
      - config/settings/{base,local,test}.py
      - config/urls.py
    - manage.py

from __future__ import absolute_import

import os
from celery import Celery, signals

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.local')

app = Celery('celery_uncovered')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app  # noqa

__all__ = ['celery_app']

__version__ = '0.0.1'
__version_info__ = tuple([int(num) if num.isdigit() else num for num in __version__.replace('-', '.', 1).split('.')])

CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='amqp://guest:guest@localhost:5672//')
CELERY_RESULT_BACKEND = 'django-db+sqlite:///results.sqlite'

@shared_task
def fetch_hot_repos(since, per_page, page):
    payload = {
        'sort': 'stars', 'order': 'desc', 'q': 'created:>={date}'.format(date=since),
        'per_page': per_page, 'page': page,
        'access_token': settings.GITHUB_OAUTH}
    headers = {'Accept': 'application/vnd.github.v3+json'}
    connect_timeout, read_timeout = 5.0, 30.0
    r = requests.get(
        'https://api.github.com/search/repositories',
        params=payload,
        headers=headers,
        timeout=(connect_timeout, read_timeout))
    items = r.json()[u'items']
    return items

@shared_task
def produce_hot_repo_report(period, ref_date=None):
    # 1. parse date
    ref_date_str = strf_date(period, ref_date=ref_date)

    # 2. fetch and join
    fetch_jobs = group([
        fetch_hot_repos.s(ref_date_str, 100, 1),
        fetch_hot_repos.s(ref_date_str, 100, 2),
        fetch_hot_repos.s(ref_date_str, 100, 3),
        fetch_hot_repos.s(ref_date_str, 100, 4),
        fetch_hot_repos.s(ref_date_str, 100, 5)
    ])
    # 3. group by language and
    # 4. create csv
    return chord(fetch_jobs)(build_report_task.s(ref_date_str)).get()


@shared_task
def build_report_task(results, ref_date):
    all_repos = []
    for repos in results:
        all_repos += [Repository(repo) for repo in repos]

    # 3. group by language
    grouped_repos = {}
    for repo in all_repos:
        if repo.language in grouped_repos:
            grouped_repos[repo.language].append(repo.name)
        else:
            grouped_repos[repo.language] = [repo.name]

    # 4. create csv
    lines = []
    for lang in sorted(grouped_repos.keys()):
        lines.append([lang] + grouped_repos[lang])

    filename = '{media}/github-hot-repos-{date}.csv'.format(media=settings.MEDIA_ROOT, date=ref_date)
    return make_csv(filename, lines)

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'produce-csv-reports': {
        'task': 'celery_uncovered.toyex.tasks.produce_hot_repo_report_task',
        'schedule': crontab(minute=0, hour=0)  # midnight,
        'args': ('today',)
    },
}

    $ celery -A celery_uncovered worker -l info

from datetime import date
from celery_uncovered.toyex.tasks import produce_hot_repo_report_task
produce_hot_repo_report_task.delay('today').get(timeout=5)

    # Start celerybeat with the following command
    $ celery -A celery_uncovered beat -l info

@shared_task
def report_error_task(subject, message, *args, **kwargs):
    mail_admins(subject, message, *args, **kwargs)

from django.utils.log import AdminEmailHandler
from celery_uncovered.handlers.tasks import report_error_task

class CeleryHandler(AdminEmailHandler):

    def send_mail(self, subject, message, *args, **kwargs):
        report_error_task.delay(subject, message, *args, **kwargs)

LOGGING = {
   'version': 1,
   'disable_existing_loggers': False,
   ...,
   'handlers': {
       ...
       'mail_admins': {
           'level': 'ERROR',
           'filters': ['require_debug_true'],
           'class': 'celery_uncovered.toyex.log_handlers.admin_email.CeleryHandler'
       }
   },
   'loggers': {
       'django': {
           'handlers': ['console', 'mail_admins'],
           'level': 'INFO',
       },
       ...
   }
}

    $ docker build . -f docker/mailhog/Dockerfile -t mailhog/mailhog:latest
    $ docker run -d -p 1025:1025 -p 8025:8025 mailhog/mailhog
    $ # navigate with your browser to localhost:8025

EMAIL_BACKEND = env('DJANGO_EMAIL_BACKEND', default='django.core.mail.backends.smtp.EmailBackend')
EMAIL_PORT = 1025
EMAIL_HOST = env('EMAIL_HOST', default='mailhog')

@signals.celeryd_after_setup.connect
def configure_task_logging(instance=None, **kwargs):
    tasks = instance.app.tasks.keys()
    LOGS_DIR = settings.ROOT_DIR.path('logs')

if not os.path.exists(str(LOGS_DIR)):
        os.makedirs(str(LOGS_DIR))
        print 'dir created'
    default_handler = {
        'level': 'DEBUG',
        'filters': None,
        'class': 'logging.FileHandler',
        'filename': ''
    }
    default_logger = {
        'handlers': [],
        'level': 'DEBUG',
        'propogate': True
    }
    LOG_CONFIG = {
        'version': 1,
        # 'incremental': True,
        'disable_existing_loggers': False,
        'handlers': {},
        'loggers': {}
    }
    for task in tasks:
        task = str(task)
        if not task.startswith('celery_uncovered.'):
            continue
        task_handler = copy_dict(default_handler)
        task_handler['filename'] = str(LOGS_DIR.path(task + ".log"))

task_logger = copy_dict(default_logger)
        task_logger['handlers'] = [task]

LOG_CONFIG['handlers'][task] = task_handler
        LOG_CONFIG['loggers'][task] = task_logger
    logging.config.dictConfig(LOG_CONFIG)

import celery_uncovered.tricks.celery_conf

class LoggingTask(current_app.Task):
    abstract = True
    ignore_result = False

    @property
    def logger(self):
        logger = get_task_logger(self.name)
        return logger

    def log_msg(self, msg, *msg_args):
        self.logger.debug(msg, *msg_args)

    $ celery -A celery_uncovered worker -l info

from datetime import date
from celery_uncovered.tricks.tasks import add
add.delay(1, 3)

    Result of 1 + 2 = 3
    Result of 1 + 2 = 3
    ...

class ScopeBasedTask(current_app.Task):
    abstract = True
    ignore_result = False
    default_locale_id = DEFAULT_LOCALE_ID
    scope_args = ('locale_id',)

    def __init__(self, *args, **kwargs):
        super(ScopeBasedTask, self).__init__(*args, **kwargs)
        self.set_locale(locale=kwargs.get('locale_id',
            None))

    def set_locale(self, scenario_id=None):
        self.locale_id = self.default_locale_id
        if locale_id:
            self.locale_id = locale_id
        else:
            self.locale_id = get_current_locale().id

    def apply_async(self, args=None, kwargs=None, **other_kwargs):
        self.inject_scope_args(kwargs)
        return super(ScopeBasedTask, self).apply_async(args=args,
            kwargs=kwargs, **other_kwargs)

    def __call__(self, *args, **kwargs):
        task_rv = super(ScopeBasedTask, self).__call__(*args,
            **kwargs)
        return task_rv

    def inject_scope_args(self, kwargs):
        for arg in self.scope_args:
            if arg not in kwargs:
                kwargs[arg] = getattr(self, arg)

@shared_task(bind=True, base=ScopeBasedTask)
def read_scenario_file_task(self, **kwargs):
    fixture_parts = ["locales", "sc_%i.json" %
         kwargs['scenario_id']]
    return read_fixture(*fixture_parts)