Django+Celery でバッチスケジュール

##概要

• Django で開発したアプリケーションでバッチスケジュールを行う。

• Celery を利用して開発する。

  ※詳細は学習し切れていないのでこれから勉強。動かすところまでメモ。

##やりたいこと

• Django で開発したアプリケーションの裏で、バッチ処理をスケジューリングする
• アプリケーションと同一サーバ上で処理を起動する

環境

• Mac OS X Yosemite 10.10.4
• Python3.4
• Django1.8
• Celery3.1
• django-celery3.1

参考情報

• Celery
  • http://celery.readthedocs.org/en/latest/index.html
• Periodic Tasks
  • http://celery.readthedocs.org/en/latest/userguide/periodic-tasks.html

##準備

• 必要なパッケージを pip install

pip install celery
pip install django-celery

##コーディング

• 構成
  • sebastian(Project)
  • appmanager(Application)

sebastian
  └-appmanager
      └-tasks.py
      └-
     
  └-sebastian
      └-celery.py
      └-settings.py
      

• コード

sebastian/settings.py

# 30秒毎に処理起動(Cronのように日時指定の設定も可能）
from datetime import timedelta
CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'appmanager.tasks.test_celery',
        'schedule': timedelta(seconds=30),
        'args': (16, 16)
    },
}

CELERY_TIMEZONE = 'UTC'

BROKER_URL = 'django://'

sebastian/celery.py

from __future__ import absolute_import

import os

from celery import Celery

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

from django.conf import settings

app = Celery('sebastian')

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

app.conf.update(
    CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend',
)

app.conf.update(
    CELERY_RESULT_BACKEND='djcelery.backends.cache:CacheBackend',
)

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

appmanager/tasks.py

from __future__ import absolute_import

import logging
from celery import task
from celery.utils.log import get_task_logger
from appmanager import models

@task
def test_celery(x, y):
    logger = get_task_logger(__name__)
    # orm で取得した結果をログ出力し続ける(Django のログでなく、Celeryのログ）
    application_inf = models.ApplicationInf.objects.get(pk=8)
    logger.info('func start  ----------------->')
    logger.info('application:%s', application_inf.name)
    logger.info('func end -------------------->')
    return x + y
    

• 実行

celery -A sebastian worker -B -l info

• 実行結果

func start  ----------------->
application:TEST_APP
func end -------------------->