django-celeryを使わずにDjangoとCeleryを組み合わせて使う

Python Django Celery

DjangoとCeleryを組み合わせて使う際に、django-celeryのモデルのテーブルを作りたくなかったり、各種機能が不要でシンプルに動かしたかった。
Celery - Distributed Task Queue — Celery 5.3.0b2 documentation
Celeryやdjango-celeryはドキュメントが少し足りなくて、設定方法がわかりにくかったのでメモを残しておく。
自分の都合の良いように設定してるので、よくわからない人はdjango-celeryを使ったほうがいいと思う。
試したバージョンは、Python3.4, Django1.8, Celery3.1。

Celeryのアプリケーションを返す関数の用意

まず、Celeryのインスタンスを返す関数を用意する。これはCelery起動用に使うのと、Django側のセットアップに使う。
Djangoのプロジェクト名はmyproject。Celeryの設定は、 django.conf.settings を読むことにする。

myproject/celery_app.py
app = None


def get_celery_app():
    global app
    import celery
    if app is not None:
        return app
    app = celery.Celery('myproject')
    app.config_from_object('django.conf:settings')
    return app

Celeryを起動するためのモジュールと設定を用意する

celeryコマンドには、celery.appという名前でアクセス可能なモジュールを指定するので、プロジェクトディレクトリに用意する。
作成しておいたget_celery_app関数を呼ぶ前に、django.setupでDjangoのアプリケーションをロードしておくことに注意する(Djangoアプリの初期化処理などを動かしておくため)。
autodiscover_tasksメソッドで各Djangoアプリケーションのtasks.pyをロードする。

myproject/celery.py
import os

import django
from django.conf import settings

from .celery_app import get_celery_app

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()

app = get_celery_app()
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
myproject/settings.py
# (省略)

# Celeryの設定を追記
CELERY_ALWAYS_EAGER = False  # 開発時はEAGERで動かすと便利
CELERY_EAGER_PROPAGATES_EXCEPTIONS = True
BROKER_URL = 'redis://'  # BrokerにはとりあえずRedisを使っとく

ここまでファイルを用意すると、以下のコマンドでCeleryのワーカーを起動できる。ブローカーを事前に起動しておくことに注意(この設定だとRedis)

$ celery -A myproject worker

Djangoの設定ファイルを別のものに切り替えたい場合は、DJANGO_SETTINGS_MODULE環境変数に指定することで変更できる。

Django側でCeleryに設定を反映するためのアプリケーションを用意する

AppConfigのreadyでget_celery_app関数を呼んで、Djangoのsettings.pyに記載したCeleryの設定を適用するためのクラスを用意する。

celery_config/apps.py
from django.apps import AppConfig


class CeleryConfig(AppConfig):
    name = 'celery_config'

    def ready(self):
        from myproject.celery_app import get_celery_app
        get_celery_app()
celery_config/__init__.py

用意したクラスは、__init__.pyでデフォルトに設定しておく。

default_app_config = 'celery_config.apps.CeleryConfig'
myproject/settings.py(抜粋)

celery_configを有効にする

INSTALLED_APPS = (
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'celery_config',  # 追加した
    'myapp',
)

タスクを実装する

CeleryのタスクをDjangoアプリケーション内のtasks.pyに記述する。

myapp/tasks.py
from celery import shared_task

from .models import Message


@shared_task
def create_message(body):
    Message.objects.create(body=body)

Djangoアプリケーションからタスクを呼ぶ

タスクを呼ぶには、tasks.pyの関数をdelayメソッドで呼べばよい。

myapp/views.py
def create_message(request):
    tasks.create_message.delay(request.GET.get('body'))
    return render(request, 'create_message.html')

終わり。

コード

この設定例のプロジェクトのコードはgithubに置いた。
sample_nullpobug/django/django_celery at main · tokibito/sample_nullpobug · GitHub