Django Signals を使ってコードを簡素化、デカップリングする。

システムは時代とともに複雑化しており、システムのデカップリングが必要になっています。デカップリングはシステムの複雑さを軽減するだけでなく、システムの各部分を個別に管理できるため、長期的にはシステムの構築、拡張、保守が容易になります。デカップリングされたシステムでは、故障したコンポーネントがシステム全体を引きずり込むことがないため、耐故障性も向上しています。

Django は強力なオープンソースの Web フレームワークで、小さなシステムだけでなく、大規模で複雑なシステムの構築にも使用できます。モデル-テンプレート-ビューのアーキテクチャパターンに従っており、開発者が複雑なデータ駆動型のWebベースのアプリケーションの配信を実現するのを支援するという目標に忠実なのです。

Django は、1つのプロジェクト内に別々のアプリを構築することで、システムの機能を切り離すことを可能にします。例えば、ショッピングシステムがあり、アカウント、レシートのメール送信、通知などを扱う別々のアプリを持つことができます。

このようなシステムでは、特定のイベントが発生したときに、複数のアプリがアクションを実行する必要がある場合があります。あるイベントは、お客様が注文をしたときに発生することがあります。例えば、ユーザーに電子メールで通知し、サプライヤーやベンダーに注文を送信すると同時に、支払いを受け取り、処理することができるようにする必要があります。これらのイベントは全て同時に発生し、アプリケーションは分離されているので、全てのコンポーネントを同期させる必要がありますが、どうやってこれを実現するのでしょうか？

Django Signals はこのような状況で便利です。必要なことは、ユーザが注文したときにシグナルを送信し、関連または影響を受ける全てのコンポーネントがそれを聞き取り、処理を実行することです。この投稿でシグナルについてもっと調べてみましょう。

シグナルの概要

Django の Signals は Observer パターンの実装です。このようなデザインパターンでは、複数のオブジェクトが特定のオブジェクトとそれに起こるかもしれないイベントを購読、つまり「観察」する購読メカニズムが実装されています。例えば、YouTubeのチャンネルを購読している人は、コンテンツ制作者が新しいコンテンツをアップロードすると、その通知を受け取ります。

シグナルディスパッチャ」を通して、 Django は様々なシステムコンポーネントに登録された「レシーバ」にシグナルを非連続的に配信することができます。シグナルは登録され、特定のイベントが発生するたびにトリガされ、そのイベントのリスナーはイベントが発生したことを通知され、同時に受信者の機能に関連する可能性のあるペイロード内のいくつかのコンテキストデータを受け取ります。receiverはPythonの関数やメソッドであれば何でも構いません。これについては後で詳しく説明します。

シグナルディスパッチャの他に、 Django にはいくつかの便利なシグナルが同梱されており、それをリッスンすることができます。それらは以下の通りです。

post_save` は、新しい Django モデルが作成され、保存されたときに送られます。例えば、ユーザがサインアップしたときや、新しい投稿をアップロードしたときなどです。
pre_delete: Django モデルが削除される直前に送られます。良いシナリオは、ユーザがメッセージやアカウントを削除するときでしょう。
Django が HTTP リクエストの処理を完了するたびに発行されます。これは、ウェブサイトを開いたり、特定のリソースにアクセスしたりすることから始まります。

Django のもう一つの利点は、高度にカスタマイズ可能なフレームワークであることです。私たちの場合、カスタムシグナルを作成し、組み込みのシステムを使って、デカップリングされたシステムでシグナルをディスパッチしたり、受信したりすることができます。デモセクションでは、 Django の組み込みシグナルのいくつかにサブスクライブし、また、独自のカスタムシグナルをいくつか作成する予定です。

しかし、その前に Django のシグナルを利用した簡単な例を見てみましょう。ここでは、2つの関数が互いにピンポンしていますが、シグナルを通してやりとりしています。

from django.dispatch import Signal, receiver


# Create a custom signal
ping_signal = Signal(providing_args=["context"])


class SignalDemo(object):
    # function to send the signal
    def ping(self):
        print('PING')
        ping_signal.send(sender=self.__class__, PING=True)


# Function to receive the signal
@receiver(ping_signal)
def pong(**kwargs):
    if kwargs['PING']:
        print('PONG')


demo = SignalDemo()
demo.ping()

この単純なスクリプトでは、シグナルを送信するメソッドを持つクラスと、受信と応答を行う別の関数をクラスの外に作りました。この例では、シグナルの送信者はシグナルと一緒に PING コマンドを送信し、受信関数は PING コマンドが存在するかどうかをチェックし、それに対する応答として PONG を出力します。シグナルは Django の Signal クラスで作成され、 @receiver デコレータを持つ任意の関数で受信されます。

スクリプトの出力です。

$ python signal_demo.py


PING
PONG

通常は、 ping() 関数の中から pong() 関数を呼び出す必要がありますが、シグナルを使うことで、似たような、しかし分離された解決策を得ることができるのです。pong()関数は別のファイルプロジェクトに存在しても、PING`シグナルに応答することができるようになりました。

シグナルを使用する場合

Django シグナルが何であり、どのように動作するかを既に確認しましたが、他のフレームワークの機能と同様に、あらゆる場面で使うべきものではありません。Django シグナルを使うことが強く推奨される特定のシナリオがあり、それは以下の通りです。

同じイベントに興味を持つ多くの別々のコード片がある場合、シグナルはイベント通知を分散させるのに役に立ちますが、同じ時点で全ての異なるコード片を起動するのは、整頓されず、バグを誘発する可能性があります。
Django シグナルは、RESTful 通信メカニズムによる相互作用の代わりに、非結合システムのコンポーネント間の相互作用を扱うために使うこともできます。
サードパーティのライブラリを拡張する際、その変更は避けたいが、追加機能を追加する必要がある場合、シグナルは役に立ちます。

シグナルの利点

Django のシグナルは、様々な方法で非結合システムの実装を単純化します。再利用可能なアプリケーションの実装を助け、機能を何度も再実装したり、システムの他の部分を修正したりする代わりに、他のコードに影響を与えずにシグナルに応答するだけでよいのです。このように、システムのコンポーネントは、既存のコードベースに手を加えることなく、修正、追加、削除することができます。

また、シグナルは、非結合型システムの異なるコンポーネントを互いに同期させ、最新の状態に保つための簡便なメカニズムも提供する。

デモプロジェクト

このデモプロジェクトでは、ユーザーがサイトにアクセスし、利用可能な求人を閲覧し、求人情報を選んで購読するシンプルな求人掲示板を構築します。ユーザーはメールアドレスを送信するだけで購読でき、求人に変更があった場合は通知されます。例えば、条件の変更、募集の終了、求人広告の削除などです。これらの変更はすべて管理者が行い、管理者はダッシュボードで求人情報の作成、更新、削除を行うことができます。

アプリケーションを切り離すという意味で、メインのJobs Boardアプリケーションと、必要な時にユーザーに通知するためのNotificationsアプリケーションを別々に構築することにします。そして、メインの Jobs Board アプリから Notifications アプリの機能を呼び出すためにシグナルを使用します。

Django の豊富な機能セットのもう一つの証は、管理者が仕事を管理するために使うビルトインの管理ダッシュボードです。そのための作業は大幅に削減され、アプリケーションのプロトタイプをより速く作成することができます。

プロジェクトセットアップ

Pythonのプロジェクトは、システムのPythonのセットアップに影響を与えないように、隔離された環境で作業するために、仮想環境で構築するのが良い習慣です。

まずは環境を整えましょう。

# Set up the environment
$ pipenv install --three


# Activate the virtual environment
$ pipenv shell


# Install Django
$ pipenv install django

Django には、プロジェクトの作成、アプリの作成、データの移行、コードのテストなど、様々なタスクを実行するのに役立つコマンドが付属しています。プロジェクトを作成します。

# Create the project
$ django-admin startproject jobs_board &amp;&amp; cd jobs_board


# Create the decoupled applications
$ django-admin startapp jobs_board_main
$ django-admin startapp jobs_board_notifications

上のコマンドは Django プロジェクトを作成し、その中に二つのアプリケーションを作成します。セットアップが成功したことを確認するために、 Django に付属するデフォルトの migrations を移行して、データベースとテーブルをセットアップしてみましょう。

$ python manage.py migrate
$ python manage.py runserver

Django プロジェクトのローカルで動作しているインスタンスにアクセスすると、以下のように表示されるはずです。

これは、Django プロジェクトのセットアップが成功し、ロジックの実装を開始できるようになったことを意味します。

実装

Django はモデル・ビュー・テンプレートのアーキテクチャパターンに基づいており、このパターンが私たちの実装の指針にもなります。私たちはデータを定義するモデルを作成し、データへのアクセスや操作を行うビューを実装し、最後にブラウザ上のエンドユーザにデータをレンダリングするテンプレートを作成します。

私たちのアプリケーションをメインの Django アプリケーションに統合するには、以下のように INSTALLED_APPS の下にある jobs_board/settings.py に追加する必要があります。

INSTALLED_APPS = [
    # Existing apps remain...


# jobs_board apps
    'jobs_board_main',
    'jobs_board_notifications',
]

パート1：メインジョブボードアプリ

このアプリはシステムの機能の大部分を占め、ユーザーとのインタラクションポイントとなります。このアプリにはモデル、ビュー、テンプレート、そしてNotificationsアプリとやり取りするための特別なシグナルが含まれます。

まず、jobs_board_main/models.pyでモデルを作成します。

# jobs_board_main/models.py


class Job(models.Model):
    company = models.CharField(max_length=255, blank=False)
    company_email = models.CharField(max_length=255, blank=False)
    title = models.CharField(max_length=255, blank=False)
    details = models.CharField(max_length=255, blank=True)
    status = models.BooleanField(default=True)
    date_created = models.DateTimeField(auto_now_add=True)
    date_modified = models.DateTimeField(auto_now=True)


class Subscriber(models.Model):
    email = models.CharField(max_length=255, blank=False, unique=True)
    date_created = models.DateTimeField(auto_now_add=True)
    date_modified = models.DateTimeField(auto_now=True)


class Subscription(models.Model):
    email = models.CharField(max_length=255, blank=False, unique=True)
    user = models.ForeignKey(Subscriber, related_name="subscriptions", on_delete=models.CASCADE)
    job = models.ForeignKey(Job, related_name="jobs", on_delete=models.CASCADE)
    date_created = models.DateTimeField(auto_now_add=True)
    date_modified = models.DateTimeField(auto_now=True)

求人情報を定義するモデルを作成します。このモデルは、会社名と求人情報の詳細、および求人情報のステータスのみを持つことになります。また、メールアドレスだけを持っている購読者を保存するモデルも用意します。購読者と求人はSubscription` モデルを通して一緒になり、求人広告の購読に関する詳細が保存されます。

モデルを配置したら、データベースにテーブルを作成するためにマイグレーションを行う必要があります。

$ python manage.py makemigrations
$ python manage.py migrate

次に、アプリケーションのビューセクションに進みます。すべての求人情報を表示するビューと、ユーザーがメールを送信することで購読できる個別の求人情報を表示するビューを作成しましょう。

まず、すべての求人情報を表示するためのビューを作成します。

# jobs_board_main/views.py


from .models import Job


def get_jobs(request):
    # get all jobs from the DB
    jobs = Job.objects.all()
    return render(request, 'jobs.html', {'jobs': jobs})

このプロジェクトでは、関数ベースのビューを使用します。クラスベースのビューという選択肢もありますが、ここでは触れません。すべての求人情報をデータベースに問い合わせ、求人情報を表示するテンプレートを指定してリクエストに応答し、さらに求人情報をレスポンスに含めます。

Django には Jinja テンプレートエンジンが付属しており、エンドユーザに表示される HTML ファイルを作成するために使用します。jobs_board_mainアプリケーションでは、templates` フォルダを作成し、エンドユーザーにレンダリングするすべての HTML ファイルをホストします。

すべての求人情報を表示するテンプレートは、次のように個々の求人情報へのリンクがあるすべての求人情報を表示します。

<!-- jobs_board_main/templates/jobs.html --
<!DOCTYPE html

<html
<head
<titleJobs Board Homepage</title
</head
<body
<h2 Welcome to the Jobs board </h2


{% for job in jobs %}
      <div
<a href="/jobs/{{ job.id }}"{{ job.title }} at {{ job.company }}</a
<p
          {{ job.details }}
        </p
</div
    {% endfor %}


</body
</html

求人情報モデル、すべてのビューを取得して表示するget_jobsビュー、そして求人情報リストをレンダリングするテンプレートを作成しました。これら全ての作業をまとめるために、求人情報にアクセスできるエンドポイントを作成する必要があります。これはjobs_board_main_applicationにurls.py` ファイルを作成することで行います。

# jobs_board_main/urls.py


from django.urls import path
from .views import get_jobs


urlpatterns = [
    # All jobs
    path('jobs/', get_jobs, name="jobs_view"),
]

このファイルでは、ビューをインポートし、パスを作成し、ビューをそれにアタッチします。次に、アプリケーションのURLを jobs_board プロジェクトフォルダ内の urls.py ファイルに登録します。

# jobs_board/urls.py


urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('jobs_board_main.urls')), # &lt;--- Add this line
]

これで、プロジェクトをテストする準備ができました。アプリケーションを実行して、localhost:8000/jobsに移動すると、このような表示になります。

現在、ジョブはありません。Django には、データ入力を行うために使用できる管理アプリケーションが同梱されています。まず、スーパーユーザを作成するところから始めます。

スーパーユーザを作成したら、 jobs_board_main アプリケーションの admin.py ファイルにモデルを登録する必要があります。

# jobs_board_main/admin.py
from django.contrib import admin
from .models import Job


# Register your models here.
admin.site.register(Job)

アプリケーションを再起動し、localhost:8000/adminに移動して、先ほど設定した認証情報でログインします。これがその結果です。

Jobs” の行にあるプラス記号をクリックすると、求人情報の詳細を記入するフォームが表示されます。

求人を保存して jobs エンドポイントに戻ると、先ほど作成した求人広告が表示されます。

これから、1つの求人を表示し、ユーザーがEメールを送信することで購読できるようにするためのビュー、テンプレート、URLを作成します。

jobs_board_main/views.py` は次のように拡張されます。

# jobs_board_main/views.py
# previous code remains
def get_job(request, id):
    job = Job.objects.get(pk=id)
    return render(request, 'job.html', {'job': job})


def subscribe(request, id):
    job = Job.objects.get(pk=id)
    sub = Subscriber(email=request.POST['email'])
    sub.save()


subscription = Subscription(user=sub, job=job)
    subscription.save()


payload = {
      'job': job,
      'email': request.POST['email']
    }
    return render(request, 'subscribed.html', {'payload': payload})

また、求人広告の単一ビューのテンプレートを templates/job.html に作成する必要があります。このテンプレートには、ユーザーの電子メールを取り込んで求人広告を購読するためのフォームが含まれています。

<!-- jobs_board_main/templates/job.html --
<html
<head
<titleJobs Board - {{ job.title }}</title
</head
<body
<div
<h3{{ job.title }} at {{ job.company }}</h3
<p
          {{ job.details }}
        </p
<br/
<pSubscribe to this job posting by submitting your email</p
<form action="/jobs/{{ job.id }}/subscribe" method="POST"
          {% csrf_token %}
          <input id="email" name="email" placeholder="Enter your email" type="email"/
<input type="submit" value="Subscribe"/
</form
<hr/
</div
</body
</html

ユーザーが求人広告を購読したら、次のような subscribed.html テンプレートを持つ確認ページにリダイレクトさせる必要があります。

<!-- jobs_board_main/templates/subscribed.html --
<!DOCTYPE html

<html
<head
<titleJobs Board - Subscribed</title
</head
<body
<div
<h3Subscription confirmed!</h3
<p
          Dear {{ payload.email }}, thank you for subscribing to {{ payload.job.title }}
        </p
</div
</body
</html

最後に、新しい機能はエンドポイントを介して公開する必要があります。これは既存の jobs_board_main/urls.py に次のように追加します。

# jobs_board_main/urls.py
from .views import get_jobs, get_job, subscribe


urlpatterns = [
    # All jobs
    path('jobs/', get_jobs, name="jobs_view"),
    path('jobs/<int:id', get_job, name="job_view"),
    path('jobs/<int:id/subscribe', subscribe, name="subscribe_view"),
]

求人情報のリストを表示し、クリックして、更新を受け取るメールアドレスを送信することで、メインのJobs Boardアプリケーションをテストすることができます。

さて、動くアプリケーションができたので、次は Django Signals を導入して、特定のイベントが発生したときにユーザや購読者に通知する番です。求人情報は、私たちが記録した特定の会社のメールアドレスに紐づいており、新しいユーザがその求人情報を購読したら、その会社に通知したいのです。また、求人広告が削除されたときにも、購読しているユーザーに通知したいです。

求人広告が削除されたときにユーザに通知するために、 Django の組み込みシグナル post_delete を利用します。また、 new_subscriber というシグナルを作成し、ユーザーが求人広告を購読したときに企業に通知するために使用します。

カスタムシグナルの作成は、 jobs_board_main アプリケーション内に signals.py ファイルを作成することで行います。

# jobs_board_main/signals.py
from django.dispatch import Signal


new_subscriber = Signal(providing_args=["job", "subscriber"])

これで完了です。jobs_board_main/views.py`にあるように、ユーザーが求人広告の購読に成功した後に、カスタムシグナルを呼び出す準備ができました。

# jobs_board_main/views.py


# Existing imports and code are maintained and truncated for brevity
from .signals import new_subscriber


def subscribe(request, id):
    job = Job.objects.get(pk=id)
    subscriber = Subscriber(email=request.POST['email'])
    subscriber.save()


subscription = Subscription(user=subscriber, job=job, email=subscriber.email)
    subscription.save()


# Add this line that sends our custom signal
    new_subscriber.send(sender=subscription, job=job, subscriber=subscriber)


payload = {
      'job': job,
      'email': request.POST['email']
    }
    return render(request, 'subscribed.html', {'payload': payload})

Django は pre_delete シグナルを送りますので、心配する必要はありません。

パート2: ジョブズボード通知アプリ

すでに jobs_board_notifications アプリケーションを作成し、Django プロジェクトと接続しました。このセクションでは、メインのアプリケーションから送られたシグナルを消費して、通知を送信します。Django にはメールを送信する機能が組み込まれていますが、開発目的では、代わりにコンソールにメッセージを出力することにします。

私たちの jobs_board_notifications アプリケーションは、ユーザとのインタラクションを必要としないので、そのためのビューやテンプレートを作成する必要はありません。私たちの jobs_board_notifications の唯一の目標は、シグナルを受信して通知を送信することです。この機能は models.py で実装することにします。

jobs_board_notifications/models.py` でシグナルを受信してみましょう。

# jobs_board_notifications/models.py.
from django.db.models.signals import pre_delete
from django.dispatch import receiver


from jobs_board_main.signals import new_subscriber
from jobs_board_main.models import Job, Subscriber, Subscription


@receiver(new_subscriber, sender=Subscription)
def handle_new_subscription(sender, **kwargs):
    subscriber = kwargs['subscriber']
    job = kwargs['job']


message = """User {} has just subscribed to the Job {}.
    """.format(subscriber.email, job.title)


print(message)


@receiver(pre_delete, sender=Job)
def handle_deleted_job_posting(**kwargs):
    job = kwargs['instance']


# Find the subscribers list
    subscribers = Subscription.objects.filter(job=job)


for subscriber in subscribers:
        message = """Dear {}, the job posting {} by {} has been taken down.
        """.format(subscriber.email, job.title, job.company)


print(message)

jobs_board_notificationsでは、カスタムシグナルとpre_saveシグナル、そしてモデルをインポートします。receiver デコレータを使って、キーワード引数として渡されたシグナルとコンテキストデータを取得します。

コンテキストデータを受け取ると、私たちが送信したシグナルに応答して、ユーザーが購読し、求人情報が削除されたときに、購読者と会社に「メール」（単純化のためにコンソールに出力していることを思い出してください）を送信するためにそれを使用します。

テスト

管理者ダッシュボードで求人情報を作成したら、ユーザーが購読できるようになります。ユーザーが購読すると、jobs_board_notificationsアプリケーションから投稿を所有する会社に次のようなメールが送信されます。

これは new_subscriber シグナルが jobs_board_main アプリケーションから送信され、 jobs_board_notifications アプリケーションで受信されたことの証明になります。

求人広告が削除されると、次のように、その求人広告を購読していたすべてのユーザーにメールで通知されます。

Django の pre_delete シグナルは便利で、私たちのハンドラは購読しているユーザに、特定の求人広告が削除されたことを通知します。

概要

この記事では、特定のイベントに応答して Django Signals を介して通信する 2 つのアプリケーションを持つ Django プロジェクトを構築しました。この 2 つのアプリケーションは分離されており、アプリケーション間の通信の複雑さは大幅に軽減されています。ユーザが求人広告を購読すると、私たちは企業に通知します。一方、求人広告が削除されたときは、購読しているすべての顧客にその求人広告が削除されたことを通知します。

しかし、 Django Signals を利用する際に注意しなければならないことがいくつかあります。シグナルが十分に文書化されていない場合、新しいメンテナは、ある問題や予期せぬ動作の根本原因を特定するのに苦労する可能性があります。したがって、アプリケーションでシグナルを使用する場合、使用するシグナル、それを受信する場所、そしてその背後にある理由を文書化するのが良いアイデアです。これは、コードを保守する人がアプリケーションの動作を理解し、問題をより速く、より良く解決するのに役立ちます。また、シグナルは同期的に送信されることに注意するとよいでしょう。シグナルはバックグラウンドで実行されたり、非同期ジョブによって実行されたりすることはありません。

Django の Signals に関するこれらの情報とデモプロジェクトがあれば、私たちの Django Web プロジェクトで Signals の力を利用できるはずです。

このプロジェクトのソースコードは Github で公開されています。

にあります。