Django 6.0で入ったタスクフレームワーク向けのバックエンドを、これまでに3つ作っていました。
tokibito.hatenablog.com tokibito.hatenablog.com tokibito.hatenablog.com
今回はAWSの Amazon SQS をキューに使うバックエンドを作りました。
AWSでアプリケーションを動かす場合、メッセージングにはSQSを使いたいことがあるので、そのときのために作っておきました。
django-tasks-amazon-sqs
これまでの3つと同じく、仕様を起こしてClaudeCodeにコードを書かせました。
コードはGitHubに置いています。
django-tasks-amazon-sqsというパッケージでPyPIにもアップロードしておきました。
機能は次の通りです。
- Djangoのタスクフレームワークで定義したタスクをAmazon SQS経由で実行できるバックエンドを提供
- タスクの引数はSQSのメッセージ（JSON）に含める。データベースは使わないので、モデルもマイグレーションもない
- タスクの
queue_nameがそのままSQSのキュー名になる - 管理コマンド
run_sqs_tasksでワーカーを起動し、SQSからメッセージを受信して実行する run_afterによる遅延実行（最大15分）- asyncのタスクにも対応
- ワーカーは
SIGTERM/SIGINTを受けたら実行中のタスクを終えてから終了する
8月に出たDjango 6.1にも対応していて、Django 6.0と6.1、Python 3.12〜3.14でテストしています。
アーキテクチャ
sequenceDiagram
participant App as Application
participant Backend as SQSBackend
participant SQS as Amazon SQS
participant Worker as run_sqs_tasks
Note over App,Worker: Task Enqueue
App->>Backend: task.enqueue(args, kwargs)
Backend->>Backend: Validate & serialize to JSON
Backend->>SQS: SendMessage (payload, DelaySeconds)
SQS-->>Backend: MessageId
Backend-->>App: TaskResult (id, status=READY)
Note over App,Worker: Task Execution
Worker->>SQS: ReceiveMessage (long poll)
SQS-->>Worker: Message (payload, ReceiptHandle)
Worker->>Worker: Import the task, run it
alt Success
Worker->>SQS: DeleteMessage
else Failure
Note over Worker,SQS: Message left in the queue.<br/>Visible again after the visibility timeout.<br/>The redrive policy moves it to the dead-letter queue
end
Cloud Tasks版はCloud Tasksからアプリのエンドポイントを呼び出してもらう形でしたが、SQS版はワーカーがSQSにメッセージを取りに行く形です。アプリ側に実行用のエンドポイントを公開する必要はありません。
ワーカーはロングポーリング（メッセージが届くまで最大20秒待つ受信方法）でSQSからメッセージを受け取ります。
使い方はこれまでのものとほぼ同じで、settings.pyでバックエンドを指定します。
TASKS = {
"default": {
"BACKEND": "django_tasks_amazon_sqs.SQSBackend",
"QUEUES": [],
"OPTIONS": {
"AWS_REGION": "ap-northeast-1",
},
},
}
SQSのキューは事前に作っておき、ワーカーを起動します。
aws sqs create-queue --queue-name default python manage.py run_sqs_tasks --continuous
AWSの認証情報はboto3の通常の探し方に従うので、環境変数やIAMロールで設定しておけば追加の設定は不要です。
失敗したタスクの再実行
タスクが失敗した場合、ワーカーはメッセージを削除せずにキューに残します。
SQSは受信されたメッセージを可視性タイムアウト（visibility timeout、受信後に他のワーカーから見えなくなる時間）の間だけ隠しておき、削除されなければまた受信できる状態に戻します。これで失敗したタスクが再実行されます。
何回まで再実行するかは、SQSのキューのredrive policy（再配信ポリシー）で決めます。指定した回数を超えたメッセージはデッドレターキュー（処理できなかったメッセージを移しておく別のキュー）に移されます。redrive policyを設定しないと失敗したタスクが延々と再実行されるので、設定は必須だと思ってください。
Cloud Tasks版で再送制御をCloud Tasks側に任せたのと同じ考え方で、パッケージ側では再実行の回数などは管理していません。
SQSの制約
SQSの仕様上できないことは、エラーにしています。
- 遅延実行はメッセージの
DelaySecondsを使うため、15分より先のrun_afterを指定するとInvalidTaskになる- 実行時刻になるまでメッセージをキューに入れ直す実装を、将来的に追加する予定です
- SQSにはメッセージの優先度がないので、
priorityを指定したタスクもInvalidTaskになる。急ぎの処理は別のキューに分けて、専用のワーカーを動かす想定 - FIFOキューは今のバージョンでは非対応
また、実行結果はどこにも保存しないので、get_result() で結果を取得することはできません。django-database-taskやdjango-tasks-redisは結果を保存していたので、ここは違いになります。
可視性タイムアウトは、一番長く実行されるタスクよりも長くしておく必要があります。タスクの実行中にタイムアウトすると、別のワーカーが同じタスクを実行し始めてしまいます。SQSの配信は「少なくとも1回」で、同じメッセージが2回届くこともあるので、タスクは何度実行されても問題ないように作っておくのが良いです。
django-database-taskのSQS対応
django-database-taskのほうにも、SQSをメッセージングに使うバックエンド（SQSDatabaseBackend）を入れています。現状はこちらのほうがフル機能です。
タスクの引数や結果はデータベースに保存して、SQSにはtask_idだけを送ります。15分より先に実行するタスクはSQSに送らずデータベースに置いておき、ワーカーがデータベースを見に行って実行するので、SQSの遅延の上限を気にせずに使えます。
Amazon SQS Integration・tokibito/django-database-task
データベースを使わずにSQSだけで済ませたい場合はdjango-tasks-amazon-sqs、という使い分けです。
ローカルでの開発
SQS_ENDPOINT_URL で接続先を変えられるので、AWSのモックライブラリの moto をサーバーモードで起動すれば、AWSのアカウントなしで試せます。
pip install "moto[server]" moto_server -p 5000 aws --endpoint-url http://localhost:5000 sqs create-queue --queue-name default
パッケージのテストでもSQSの代わりにmotoを使っています。
READMEではLocalStackも紹介していますが、LocalStackは2026年3月からアカウント登録と認証トークンが必要になり、無料プランは非商用に限られるようになったので、手軽に使うならmotoのほうが良さそうです。
