[Django] データ更新をメールやチャットサービスで通知する [Celery]

## この記事について

Djangoの初心者向け記事です。データの更新に合わせて、更新内容をメールやチャットサービスに投稿する方法をまとめました。Celeryを使った非同期処理については後半に書きました。

## WebAPIについて

チャットサービスへの投稿には各サービスのWebAPIを利用します。
WebAPIの利用方法は各サービスで説明しています。

- [Slack：Slack での着信 WebHook の利用](https://get.slack.help/hc/ja/articles/115005265063)
- [ChatWork API ドキュメント](http://developer.chatwork.com/ja/endpoint_rooms.html)

ここではSlackの「Incoming Webhook」を使った投稿を例に使います。
事前にエンドポイントの作成が必要なので、上記リンクを見て作成してください。

## 非同期処理を使わない場合

最初に非同期処理を使わない簡単な方法の説明です。

### 修正対象ファイル

データの更新の仕組みはすでに存在するものとして、通知機能の修正分のみ記述します。

```tree
proj
  ├ proj
  │  └ settings.py         # メール送信サーバ(SMTP)とSlackエンドポイントの設定
  └ app
     ├ tasks.py            # 通知処理の実体
     ├ views.py           # クラスベースビューのform_validで通知処理を呼び出す
     └ templates
         └ app
             └ massege.txt # 通知メッセージのテンプレート
```

### パッケージ追加

- WebAPI通信用に[requests](http://requests-docs-ja.readthedocs.io/en/latest/)が必要になるのでインストールします。

```shell
pip install requests
```

参考：[Requests の使い方 (Python Library)](https://qiita.com/sqrtxx/items/49beaa3795925e7de666)



### コード内容

- proj/settings.py

メール送信サーバ(SMTP)のとSlackエンドポイントの設定
**この方法でGmailのSTMPを使って送信する場合は、Gmail上で「安全性の低いアプリの許可」を有効にすること。**
参考：https://support.google.com/accounts/answer/6010255?hl=ja

なお、Gmailに限っては専用のパッケージがあるのでそっちを使うべきかも。
参考：https://developers.google.com/gmail/api/quickstart/python

```python:proj/settings.py
EMAIL_USE_TLS = True
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_HOST_USER = 'your_account@gmail.com'
EMAIL_HOST_PASSWORD = 'PassW@rd'
EMAIL_PORT = 587
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'

# デバッグ用：以下の設定だとコンソールに出力される
# EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend'

SLACK_WEBHOOK_ENDPOINT = 'https://your_channel_endpoint'
```


- app/tasks.py

メール送信とSlack投稿処理の内容
参考：[naritoブログ：Djangoでメールを送信](https://torina.top/detail/278/)


```python:app/tasks.py

import json

import requests
from django.core.mail import send_mail
from django.template.loader import get_template

import proj.settings as settings


# メール送信
def send_email(message):
    subject = 'データ更新'
    from_address = 'from@gmail.com'
    to_address = ['to@gmail.com']
    send_mail(subject, message, from_address, to_address)


# Slack投稿(Incoming Webhook 使用)
def send_slack_message(message):
    requests.post(settings.SLACK_WEBHOOK_ENDPOINT,
                  data=json.dumps({
                      'text': message,  # 投稿するテキスト
                      'username': u'me',  # 投稿のユーザー名
                      'icon_emoji': u':ghost:',  # 投稿のプロフィール画像に入れる絵文字
                      'link_names': 1,  # メンションを有効にする
                  }))

# メールの本文を作成する
def get_message(item, action):
    template = get_template('app/message.txt')
    context = {
        "item": item, "action": action,
    }
    message = template.render(context)
    return message


# 通知処理
def send_notification(item, action):
    message = get_message(item, action)
    send_slack_message(message)
    send_email(message)

```

- app/views.py

ビュー内でtasksで定義した通知処理の呼び出しをします。

```python:app/views.py

class MyCreateView(CreateView):

    model = Item
    form_class = ItemForm
    success_url = reverse_lazy('index')

    def form_valid(self, form):

        item = form.save(commit=False)
        item.save()
        send_notification(item, '登録')

        return HttpResponseRedirect(self.success_url)


class MyUpdateView(UpdateView):

    model = Item
    form_class = ItemForm
    success_url = reverse_lazy('index')

    def form_valid(self, form):

        item = form.save(commit=False)
        saved_item = item.save()
        send_notification(saved_item, '更新')

        return HttpResponseRedirect(self.success_url)

```

- app/templates/app/massege.txt

メールの本文です。テンプレートの仕組みを使います。HTMLと同様にテンプレートタグでロジックも組めます。

```txt:app/templates/app/massege.txt
データベースが更新されました。

処理内容：{{ action }}
作成時間：{{ item.created_at }}
作成者：{{ item.created_by }}

http://localhost:8000/detail/{{ item.pk }}
```

修正後、データ更新を試してください。問題がなければ通知処理が実行されます。

## 非同期処理を使う場合

### 非同期処理を使う目的

先のコードを試してみればわかりますが、外部サービスの応答速度はそれほど速くなく、早くて1秒、GmailのSMTPなどは3～4秒かかります。その時間ユーザーが操作不能になるのでアプリケーションの利便性が下がってしまいます。

それを防ぐため、時間のかかる処理を別プロセスに依頼して、ユーザーの操作不能時間を無くすのが非同期処理です。

### 非同期処理に必要なもの

Djangoで非同期処理を実現するには **Celery** というPythonのタスクキューサービスを使います。タスクキューサービスとは時間のかかるジョブを他のプロセスやマシンで分散実行するためのアプリケーションです。

CeleryはDjangoとは別にサービス（デーモン）として起動する必要がありますが、起動設定とジョブ定義はDjangoのプロジェクト内で一緒に管理できます。

Celeryの利用には、ジョブ内容を一時保管するためのメッセージキューというアプリケーションが必要です。[RabbitMQ](https://www.rabbitmq.com/)、[Redis](https://redis.io/)、[Amazon SQS](https://docs.aws.amazon.com/ja_jp/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html)が指定されているので、開発・本番環境にいずれかを準備する必要があります。

- [Celery公式](http://docs.celeryproject.org/en/latest/index.html)
- [Celery公式 日本語訳（旧バージョン）](https://blog.ozacc.com/docs/celery/index.html)

**Celeryを利用したデータ更新通知のイメージ**

![image.png](https://qiita-image-store.s3.amazonaws.com/0/217094/a0ee8d8c-5c4e-833a-583d-c6512886a5d1.png)

**タスクキューの役割定義**

- ジョブを登録するプロセスを**Client**と呼びます。今回の例ではWebアプリのプロセスのことです。
- Clientからジョブの要求を受けてタスクキューに登録するサービスを**Broker**と呼びます。Brokerは**[メッセージキュー](https://aws.amazon.com/jp/message-queue/)**というソフトウェアで構築されます。
- Brokerのタスクキューからジョブを取得して実際に実行するプロセスを**Worker**と呼びます。

#### 参考資料

Django+Celeryについて解説した記事は多々ありますが、以下のサイトのものがもっとも簡潔かつ明快であると思いました（Celery公式サイトよりも！）。

- [How to Use Celery and RabbitMQ with Django](https://simpleisbetterthancomplex.com/tutorial/2017/08/20/how-to-use-celery-with-django.html)

英語ですが、まずはここを一読されるのが近道だと思います。Google翻訳で十分理解できます。

#### 注意事項

- 残念ながらCeleryはVersion4以降でWindows非対応となりました。ただし今のところ起動コマンドを若干修正することで問題なく動作しますし、タスクスケジューラを使ってデーモン化し本番環境で動かすこともできます。

参考：http://docs.celeryproject.org/en/latest/faq.html#does-celery-support-windows

- Celeryの過去バージョンではDjango ORMをブローカーに使うことができましたが、最新版ではできなくなっています。それに合わせパッケージ「[django-celery](https://github.com/celery/django-celery)」も不要となりました。

### 開発手順

#### １．メッセージキューの準備

まず開発環境にメッセージキューをインストールします。Amazon SQSは開発で使えないので（使ってもいいけど）、選択肢はRabbitMQかRedisになります。

MacOS、Linuxで開発する場合はどちらでもOKです。Redisの方がメジャーなサービスのようです。ただしWindowsに関してはRedisの最新版の提供が2016年をもって終了しています。そのためRabbitMQ一択となります。

これ以降はRabbitMQでの説明となります。
RabbitMQ の構築方法については以下の記事にわかりやすくまとまっているのでこの通りに進めればOKです。

- [RabbitMQのインストールと管理画面の有効化(MacOSX/Linux/Windows)](https://qiita.com/AKB428/items/a2662fbb624ce7659025)

#### ２．コード作成


##### 修正対象ファイル

データの更新の仕組みはすでに存在するものとして、通知機能の修正分のみ記述します。

```tree
proj
  ├ proj
  │  ├ __init__.py         # Django起動時にcelery.pyのインスタンスが生成されるようにする。
  │  ├ celery.py           # celery.pyの起動ファイル
  │  └ settings.py         # メール送信サーバ(SMTP)の設定
  └ app
     ├ tasks.py            # 通知処理の実体
     ├ models.py           # save, delete メソッドに通知処理を追加
     └ templates
         └ app
             └ massege.txt # 通知メッセージのテンプレート
```

##### コード

Djangoの起動時にCeleryのインスタンスを作成します。これが無いとshared_taskデコレータが機能しません。

- `proj/__init__.py`

```python:proj/__init__.py

from .celery import app as celery_app

__all__ = ['celery_app']
```

- `proj/celery.py`

`app.config_from_object`により、Celeryの設定をDjangoの設定ファイルより取得します。`app.autodiscover_tasks()`により、各アプリケーションフォルダにある`tasks.py`をジョブ定義ファイルとして認識します。

```python:proj/celery.py

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
```

- `proj/settings.py`

「非同期処理を使わない場合」のメールサーバの設定に加えて、ブローカーのURLを指定します。Redisを使う場合は`redis://localhost`となります。

```python:proj/settings.py
CELERY_BROKER_URL = 'amqp://localhost'
```

- `app/tasks.py`

一応全体を載せますが、非同期処理を使わない場合のものとほとんど変わりません。違いは非同期タスクに`shared_task`デコレータが付いたこと、非同期タスクの呼び出しにdelayメソッドを使っていることだけです。

ちなみに非同期タスクはクラスオブジェクトを引数に指定することはできないので注意してください。

```app/tasks.py
import json

import requests
from celery import shared_task
from django.core.mail import send_mail
from django.template.loader import get_template

import proj.settings as settings


# メール送信
@shared_task
def send_email(message):
    subject = 'データ更新'
    from_address = 'from@gmail.com'
    to_address = ['to@gmail.com']
    send_mail(subject, message, from_address, to_address)


# Slack投稿(Incoming Webhook)
@shared_task
def send_slack_message(message):
    requests.post(settings.SLACK_WEBHOOK_ENDPOINT,
                  data=json.dumps({
                      'text': message,    # 投稿するテキスト
                      'username': u'me',  # 投稿のユーザー名
                      'icon_emoji': u':ghost:',  # 投稿のプロフィール画像に入れる絵文字
                      'link_names': 1,  # メンションを有効にする
                  }))


# メールの本文を作成する
def get_message(item, action):
    template = get_template('app/message.txt')
    context = {
        "item": item, "action": action,
    }
    message = template.render(context)
    return message


# 通知処理
def send_notification(item, action):
    message = get_message(item, action)
    send_slack_message.delay(message)
    send_email.delay(message)


```

views.pyとmassege.txtについては「非同期処理を使わない場合」と一緒なので省略します。

#### ３．Celeryの起動

CeleryはDjangoのプロセスとは別に起動させます。

- 開発環境ではコンソール上で起動します。
- 本番環境ではサービス（デーモン）化させます。

##### 開発環境

Djangoを起動（manage.py）したあと、別のコンソールで以下のコマンドを実行します。

```Shell:Linux・macOS
celery -A proj worker -l info
```

```Shell:Windows
celery -A proj worker -l info -P eventlet
```
Windowsでは事前に`pip install eventlet`でeventletパッケージを追加してください。WindowsでLinux、macOSと同じコマンドを実行すると、ジョブの実行時に以下のエラーが出ます。上記コマンドはその回避策です。

```:Windows
ValueError: not enough values to unpack (expected 3, got 0)
```

参考：[[github] ValueError: not enough values to unpack (expected 3, got 0)](https://github.com/celery/celery/issues/4178)

celeryコマンドがceleryインスタンスを探し出す仕組みについては以下の記事にまとまっています。
参考：https://qiita.com/kwi/items/bd289aeff0fa5881e797
上の元ネタ：https://media.readthedocs.org/pdf/celery/latest/celery.pdf

##### 本番環境

- initdを使ったCeleryのデーモン化については本家のサイトを見てください。
 - http://docs.celeryproject.org/en/latest/userguide/daemonizing.html

- こちらのサイトにもSupervisordを使った管理方法が記載しています。
 - [How to Use Celery and RabbitMQ with Django](https://simpleisbetterthancomplex.com/tutorial/2017/08/20/how-to-use-celery-with-django.html)

- Windowsについてはタスクスケジューラを使った稼働方法が推奨されていました（現在のバージョンからそもそもWindows非対応）。
 - https://www.calazan.com/windows-tip-run-applications-in-the-background-using-task-scheduler/

##### Herokuの場合

HerokuではRabbitMQ(RabbitMG-bigwig)とRedisはアドオンとして提供されています。どちらも無料コースがあるので活用しましょう。RabbitMQの場合だと環境変数`RABBITMQ_BIGWIG_URL`にブローカーのURLが格納されます。os.environやdjango-environを使って取得してください。


`Procfile`にCeleryの起動コマンドを指定します。workerプロセスとして実行します。

```shell:Procfile
web: gunicorn proj.wsgi --log-file -
worker: celery worker -A proj -l INFO
```

デプロイ直後はworkerプロセスが起動していません。以下のコマンドで登録したプロセスを起動します。

```shell
heroku ps:scale worker=1
```


参考：[Django-Celery-Redisのherokuへのデプロイがうまくできたときのメモ](https://qiita.com/k_mawa82/items/67e052531d05a39470e8)

### Celery対応パッケージ

Django+Celery環境をより使いやすくするためのパッケージが出ています。

- [django-celery-results](
http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html#django-celery-results-using-the-django-orm-cache-as-a-result-backend)
 - ジョブの実行結果をDjangoのデータベースに保存
- [django-celery-email](https://github.com/pmclanahan/django-celery-email)
 - Celeryを使ったメールの送信が簡単になる。
- [django-slack](https://django-slack.readthedocs.io/)
 - Slackのライブラリ。バックエンドにCeleryを使うことができる。