概要

SQSとS3とLambdaを使った非同期処理をローカル上でLocalStackを用いて再現する方法について解説します

前提

Djangoのアプリケーションを作成済み
SQSとLambdaとS3に関する基本的な知識を有している

今回実装する非同期処理の仕組み

①APIを使ってS3にCSVをアップロードします
②アップロードする際にSQSを使ってアップロード履歴のIDを送ります
③SQSのメッセージ送信をトリガーにLambdaを実行し、Lambda内でS3内のCSVのデータをDBにInsertするAPIを実行させます
④SQSのメッセージ内のIDを利用して該当するアップロード履歴テープル内のFileFieldを参照してアップロード処理を実行します

CSVの処理状況についてはアップロード履歴テーブル内にステータスのカラムを作成して履歴一覧から進捗状況を把握できるようにします
処理の流れは下記の画像の通りです

DjangoのCeleryを使用すれば同様のことができますが

CeleryとCelery Beat用のECSタスクを常時起動させ続ける必要があるためコストを節約したい
1週間に1回あるか、ないか程度の頻度しか使用しない

などのニーズを満たすのであれば今回の非同期処理の採用を検討してもいいのではないかと考えています

ディレクトリ構成

├── backend
│   ├── .vscode
│   │   └── launch.json
│   ├── application
│   │   ├── __init__.py
│   │   ├── admin.py
│   │   ├── apps.py
│   │   ├── fixtures
│   │   │   └── fixture.json
│   │   ├── migrations
│   │   │   ├── 0001_initial.py
│   │   │   └── __init__.py
│   │   ├── models.py
│   │   ├── serializers.py
│   │   ├── urls.py
│   │   ├── utils
│   │   │   ├── fields.py
│   │   │   ├── sqs.py
│   │   │   └── storages.py
│   │   └── views.py
│   ├── manage.py
│   ├── poetry.lock
│   ├── project
│   │   ├── __init__.py
│   │   ├── asgi.py
│   │   ├── settings.py
│   │   ├── upload
│   │   ├── urls.py
│   │   └── wsgi.py
│   └── pyproject.toml
├── containers
│   ├── django
│   │   └── Dockerfile
│   ├── localstack
│   │   ├── entrypoint.sh
│   │   ├── function.py
│   │   └── role.json
└── docker-compose.yml

LocalStackの設定

ローカル上で非同期処理を再現するためのLocalStackの設定を行います

docker-compose.yml

LocalStackではready.d配下にエントリーポイントとなるシェルスクリプトなどを配置すると起動時に実行される仕様なので下記のようにvolumeを定義します

./containers/localstack/:/etc/localstack/init/ready.d/

詳細は以下の通りです

LocalStack has four well-known lifecycle phases or stages:
BOOT: the container is running but the LocalStack runtime has not been started
START: the Python process is running and the LocalStack runtime is starting
READY: LocalStack is ready to serve requests
SHUTDOWN: LocalStack is shutting down
You can hook into each of these lifecycle phases using custom shell or Python scripts. Each lifecycle phase has its own directory in /etc/localstack/init. You can mount individual files, stage directories, or the entire init directory from your host into the container.

/etc
└── localstack
    └── init
        ├── boot.d           <-- executed in the container before localstack starts
        ├── ready.d          <-- executed when localstack becomes ready
        ├── shutdown.d       <-- executed when localstack shuts down
        └── start.d          <-- executed when localstack starts up

docker-compose.yml

version: '3.9'

services:
  db:
    container_name: db
    build:
      context: .
      dockerfile: containers/postgres/Dockerfile
    volumes:
      - db_data:/var/lib/postgresql/data
    healthcheck:
      test: pg_isready -U "${POSTGRES_USER:-postgres}" || exit 1
      interval: 10s
      timeout: 5s
      retries: 5
    environment:
      - POSTGRES_NAME
      - POSTGRES_USER
      - POSTGRES_PASSWORD
    ports:
      - '5432:5432' # デバッグ用
  app:
    container_name: app
    build:
      context: .
      dockerfile: containers/django/Dockerfile
    volumes:
      - ./backend:/code
      - ./static:/static
    ports:
      - '8000:8000'
      # デバッグ用ポート
      - '8080:8080'
    command: sh -c "/usr/local/bin/entrypoint.sh"
    stdin_open: true
    tty: true
    env_file:
      - .env
    depends_on:
      db:
        condition: service_healthy
  nginx:
    container_name: web
    build:
      context: .
      dockerfile: containers/nginx/Dockerfile
    volumes:
      - ./static:/static
    ports:
      - 80:80
    depends_on:
      - app
  localstack:
    container_name: localstack
    image: localstack/localstack:3.0.2
    ports:
      - 4566:4566
    environment:
      - DOCKER_HOST=unix:///var/run/docker.sock
      - AWS_DEFAULT_REGION=ap-northeast-1
      - LAMBDA_TOKEN=test
      - AWS_ACCESS_KEY_ID=localstack
      - AWS_SECRET_ACCESS_KEY=localstack
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - ./containers/localstack/:/etc/localstack/init/ready.d/
volumes:
  db_data:
  static:
  localstack_data:
  localstack_bin:

.env

必要な環境変数を定義します
ALLOWED_HOSTSにappを入れないとLocalStack内でAPIを実行できないので注意しましょう

.env

POSTGRES_NAME=postgres
POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres
POSTGRES_HOST=db
POSTGRES_PORT=5432
ALLOWED_HOSTS=localhost 127.0.0.1 [::1] back web app"
AWS_DEFAULT_REGION_NAME=ap-northeast-1
AWS_SQS_URL=http://sqs.ap-northeast-1.localhost.localstack.cloud:4566/000000000000/queue01.fifo
AWS_SQS_ENDPOINT=http://localstack:4566
AWS_SQS_MESSAGE_GROUP=group
LAMBDA_TOKEN=test

entrypoint.sh

エントリーポイントを記載します

Lambda用のロールの作成
LambdaがSQSを実行する用のポリシーの作成
Lambdaの作成
SQSの作成
LambdaとSQSの紐付け(イベントソースマッピング)

を行ってます

下記のコマンドで実行権限を付与しないと起動しないので注意しましょう

chmod +x containers/localstack/entrypoint.sh

entrypoint.sh

#!/bin/bash

set -eu

# 設定
LOCALSTACK_HOME=/etc/localstack/init/ready.d
AWSLOCAL_ACCOUNT_ID=000000000000
ROLE_NAME=role01
FUNC_NAME=func01
QUEUE_NAME=queue01.fifo

# 実行ロールを作成
awslocal iam create-role \
--role-name ${ROLE_NAME} \
--assume-role-policy-document file://${LOCALSTACK_HOME}/role.json

awslocal iam attach-role-policy \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole \
--role-name ${ROLE_NAME}

# Lambda 関数を作成
cp -p ${LOCALSTACK_HOME}/function.py .
chmod 755 function.py
zip function.zip function.py
awslocal lambda create-function \
    --function-name ${FUNC_NAME} \
    --zip-file fileb://function.zip \
    --handler function.lambda_handler \
    --runtime python3.12 \
    --timeout 900 \
    --memory-size 1024 \
    --role arn:aws:iam::${AWSLOCAL_ACCOUNT_ID}:role/${ROLE_NAME} \
    --environment "Variables={LAMBDA_TOKEN=${LAMBDA_TOKEN}}"

# AmazonSQSキューを作成
awslocal sqs create-queue \
    --queue-name ${QUEUE_NAME} \
    --attributes FifoQueue=true,ContentBasedDeduplication=true

# イベントソースを設定
awslocal lambda create-event-source-mapping \
--function-name ${FUNC_NAME}  \
--batch-size 1 \
--event-source-arn arn:aws:sqs:${AWS_DEFAULT_REGION}:${AWSLOCAL_ACCOUNT_ID}:${QUEUE_NAME}

role.json

Lambdaがentrypoint.shで作成したIAMロールを使って実行できるよう設定します

role.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

function.py

Lambda内に処理を記載します
API作成時に詳細に説明しますがCSVをアップロードするAPIを実行するとSQSからのメッセージをトリガーに下記のLocalStack内のLambdaの処理が自動的に実行されます
実行されるのは冒頭で説明したアップロード履歴用のIDをもとにS3内のCSVファイルをDBにinsertする処理です

function.py

import json
import os
from pprint import pprint
from urllib import error, request


def lambda_handler(event, context):
    try:
        id = event["Records"][0]["body"]
        lambda_token = os.environ.get("LAMBDA_TOKEN")
        print(id)
        url = f"http://app:8000/api/customers/register_csv"
        data = {"token": lambda_token, "id": id}
        headers = {
            "Content-Type": "application/json",
        }
        req = request.Request(url, json.dumps(data).encode(), headers)
        try:
            with request.urlopen(req) as res:
                body = json.loads(res.read())
                headers = res.getheaders()
                status = res.getcode()
                pprint(headers)
                pprint(status)
                pprint(body)
        except error.HTTPError as e:
            pprint(e)
    except Exception as e:
        print(str(e))

LocalStackの起動

以下のようにLocalStackを起動できれば成功です

2024-08-12 10:54:01 LocalStack version: 3.0.2
2024-08-12 10:54:01 LocalStack Docker container id: 1f01e94f935e
2024-08-12 10:54:01 LocalStack build date: 2023-11-29
2024-08-12 10:54:01 LocalStack build git hash: 60518e11
2024-08-12 10:54:01 
2024-08-12 10:54:02 2024-08-12T01:54:02.803  INFO --- [-functhread4] hypercorn.error            : Running on https://0.0.0.0:4566 (CTRL + C to quit)
2024-08-12 10:54:02 2024-08-12T01:54:02.803  INFO --- [-functhread4] hypercorn.error            : Running on https://0.0.0.0:4566 (CTRL + C to quit)
2024-08-12 10:54:03 Ready.
2024-08-12 10:54:03 2024-08-12T01:54:03.602  INFO --- [   asgi_gw_0] localstack.request.aws     : AWS iam.CreateRole => 200
2024-08-12 10:54:03 {
2024-08-12 10:54:03     "Role": {
2024-08-12 10:54:03         "Path": "/",
2024-08-12 10:54:03         "RoleName": "role01",
2024-08-12 10:54:03         "RoleId": "AROAQAAAAAAAONF5D6ACL",
2024-08-12 10:54:03         "Arn": "arn:aws:iam::000000000000:role/role01",
2024-08-12 10:54:03         "CreateDate": "2024-08-12T01:54:03.594000Z",
2024-08-12 10:54:03         "AssumeRolePolicyDocument": {
2024-08-12 10:54:03             "Version": "2012-10-17",
2024-08-12 10:54:03             "Statement": [
2024-08-12 10:54:03                 {
2024-08-12 10:54:03                     "Effect": "Allow",
2024-08-12 10:54:03                     "Principal": {
2024-08-12 10:54:03                         "Service": "lambda.amazonaws.com"
2024-08-12 10:54:03                     },
2024-08-12 10:54:03                     "Action": "sts:AssumeRole"
2024-08-12 10:54:03                 }
2024-08-12 10:54:03             ]
2024-08-12 10:54:03         }
2024-08-12 10:54:03     }
2024-08-12 10:54:03 }
2024-08-12 10:54:04 2024-08-12T01:54:04.075  INFO --- [   asgi_gw_0] localstack.request.aws     : AWS iam.AttachRolePolicy => 200
2024-08-12 10:54:04 updating: function.py (deflated 53%)
2024-08-12 10:54:05 2024-08-12T01:54:05.545  INFO --- [   asgi_gw_0] localstack.request.aws     : AWS lambda.CreateFunction => 201
2024-08-12 10:54:05 {
2024-08-12 10:54:05     "FunctionName": "func01",
2024-08-12 10:54:05     "FunctionArn": "arn:aws:lambda:ap-northeast-1:000000000000:function:func01",
2024-08-12 10:54:05     "Runtime": "python3.12",
2024-08-12 10:54:05     "Role": "arn:aws:iam::000000000000:role/role01",
2024-08-12 10:54:05     "Handler": "function.lambda_handler",
2024-08-12 10:54:05     "CodeSize": 591,
2024-08-12 10:54:05     "Description": "",
2024-08-12 10:54:05     "Timeout": 900,
2024-08-12 10:54:05     "MemorySize": 1024,
2024-08-12 10:54:05     "LastModified": "2024-08-12T01:54:05.542086+0000",
2024-08-12 10:54:05     "CodeSha256": "/jgoYVmWkLStYP+Ml1pxSzWykctciARBv2DiGfAKnA8=",
2024-08-12 10:54:05     "Version": "$LATEST",
2024-08-12 10:54:05     "Environment": {
2024-08-12 10:54:05         "Variables": {
2024-08-12 10:54:05             "LAMBDA_TOKEN": "test"
2024-08-12 10:54:05         }
2024-08-12 10:54:05     },
2024-08-12 10:54:05     "TracingConfig": {
2024-08-12 10:54:05         "Mode": "PassThrough"
2024-08-12 10:54:05     },
2024-08-12 10:54:05     "RevisionId": "11df305b-2c47-4dee-9372-d95c71f1fc41",
2024-08-12 10:54:05     "State": "Pending",
2024-08-12 10:54:05     "StateReason": "The function is being created.",
2024-08-12 10:54:05     "StateReasonCode": "Creating",
2024-08-12 10:54:05     "PackageType": "Zip",
2024-08-12 10:54:05     "Architectures": [
2024-08-12 10:54:05         "x86_64"
2024-08-12 10:54:05     ],
2024-08-12 10:54:05     "EphemeralStorage": {
2024-08-12 10:54:05         "Size": 512
2024-08-12 10:54:05     },
2024-08-12 10:54:05     "SnapStart": {
2024-08-12 10:54:05         "ApplyOn": "None",
2024-08-12 10:54:05         "OptimizationStatus": "Off"
2024-08-12 10:54:05     },
2024-08-12 10:54:05     "RuntimeVersionConfig": {
2024-08-12 10:54:05         "RuntimeVersionArn": "arn:aws:lambda:ap-northeast-1::runtime:8eeff65f6809a3ce81507fe733fe09b835899b99481ba22fd75b5a7338290ec1"
2024-08-12 10:54:05     }
2024-08-12 10:54:05 }
2024-08-12 10:54:06 2024-08-12T01:54:06.169  INFO --- [   asgi_gw_1] localstack.request.aws     : AWS sqs.CreateQueue => 200
2024-08-12 10:54:06 {
2024-08-12 10:54:06     "QueueUrl": "http://sqs.ap-northeast-1.localhost.localstack.cloud:4566/000000000000/queue01.fifo"
2024-08-12 10:54:06 }
2024-08-12 10:54:06 2024-08-12T01:54:06.768  INFO --- [   asgi_gw_1] localstack.request.aws     : AWS lambda.CreateEventSourceMapping => 202
2024-08-12 10:54:06 {
2024-08-12 10:54:06     "UUID": "75266f54-0fd7-4614-9342-699134681836",
2024-08-12 10:54:06     "BatchSize": 1,
2024-08-12 10:54:06     "MaximumBatchingWindowInSeconds": 0,
2024-08-12 10:54:06     "EventSourceArn": "arn:aws:sqs:ap-northeast-1:000000000000:queue01.fifo",
2024-08-12 10:54:06     "FunctionArn": "arn:aws:lambda:ap-northeast-1:000000000000:function:func01",
2024-08-12 10:54:06     "LastModified": 1723427646.764243,
2024-08-12 10:54:06     "State": "Creating",
2024-08-12 10:54:06     "StateTransitionReason": "USER_INITIATED",
2024-08-12 10:54:06     "FunctionResponseTypes": []
2024-08-12 10:54:06 }

Djangoの設定

settings.py

settings.pyにS3とSQSの設定を行います
LocalStackを使用する際はAWSのシークレットキーとアクセスキーがいるのでdocker-compose.yml内に書いたLocalStackの環境変数と同じものを記載します
逆にAWS上ではシークレットキーとアクセスキーではなく、IAMロールを使って実行するので下記のようにローカルとAWSではSQS_CLIENTの設定が違います

settings.py

import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

MEDIA_URL = "/upload/"
MEDIA_ROOT = os.path.join(BASE_DIR, "upload")

AWS_SQS_URL = os.environ.get("AWS_SQS_URL")
LAMBDA_TOKEN = os.environ.get("LAMBDA_TOKEN")

if DEBUG:
    SQS_CLIENT = boto3.client(
        "sqs",
        endpoint_url=os.environ.get("AWS_SQS_ENDPOINT"),
        aws_access_key_id="localstack",
        aws_secret_access_key="localstack",
        region_name=os.environ.get("AWS_DEFAULT_REGION_NAME"),
    )
else:
    SQS_CLIENT = boto3.client(
        "sqs",
        endpoint_url=os.environ.get("AWS_SQS_ENDPOINT"),
        region_name=os.environ.get("AWS_DEFAULT_REGION_NAME"),
    )
    DEFAULT_FILE_STORAGE = "storages.backends.s3boto3.S3Boto3Storage"
    STATICFILES_STORAGE = "storages.backends.s3boto3.S3StaticStorage"
    AWS_STORAGE_BUCKET_NAME = os.environ.get("AWS_STORAGE_BUCKET_NAME")

S3の設定

今回はローカル上でDjangoのFileSystemStorageを使用します
CustomStorageクラスを作成してローカル上ではupload/配下に、AWS上ではS3にCSVを保存するよう設定します

storages.py

from django.conf import settings
from django.core.files.storage import FileSystemStorage
from storages.backends.s3boto3 import S3Boto3Storage


if settings.DEBUG:
    storage_class = FileSystemStorage
else:
    storage_class = S3Boto3Storage


class CustomStorage(storage_class):
    file_overwrite = False
    counts = {}

    def get_alternative_name(self, file_root, file_ext):
        if file_root not in self.counts:
            self.counts[file_root] = 1
        new_file_name = (
            file_root + "_" + str(self.counts[file_root]) + file_ext
        )
        self.counts[file_root] += 1
        return new_file_name

SQSへのメッセージ送信

LocalStack、AWSでSQS_CLIENTの設定が違うのでラッパー関数を作成して下記のメソッドを実行するだけで裏側を意識せずにSQSへメッセージを送信できるようにします

sqs.py

import os

from django.conf import settings


def sqs_send_message(message_body, message_group_id=os.environ.get("AWS_SQS_MESSAGE_GROUP")):
    """AWS SQSへキューを送信する"""
    sqs_client = getattr(settings, "SQS_CLIENT")
    sqs_client.send_message(
        QueueUrl=os.environ.get("AWS_SQS_URL"),
        MessageBody=f"{message_body}",
        MessageGroupId=f"{message_group_id}",
    )

APIの実装

Modelの実装

下記のようにModelを作成します

models.py

import uuid

from django.contrib.auth.models import AbstractUser, Group
from django.contrib.auth.validators import UnicodeUsernameValidator
from django.core.validators import FileExtensionValidator, RegexValidator
from django.db import models

from application.utils.storages import CustomStorage


class User(AbstractUser):
    """システムユーザ"""

    username_validator = UnicodeUsernameValidator()

    # 不要なフィールドはNoneにすることができる
    first_name = None
    last_name = None
    date_joined = None
    groups = None
    group = models.ForeignKey(
        Group,
        on_delete=models.PROTECT,
        related_name="users",
        db_comment="システム利用者権限テーブルの外部キー",
    )
    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
        db_comment="システムユーザID",
    )
    employee_number = models.CharField(
        unique=True,
        validators=[RegexValidator(r"^[0-9]{8}$")],
        max_length=8,
        db_comment="社員番号",
    )
    username = models.CharField(
        max_length=150,
        unique=True,
        validators=[username_validator],
        db_comment="ユーザ名",
    )
    email = models.EmailField(
        max_length=254,
        unique=True,
        db_comment="メールアドレス",
    )
    created_at = models.DateTimeField(
        auto_now_add=True,
        db_comment="作成日",
    )
    updated_at = models.DateTimeField(
        auto_now=True,
        db_comment="更新日",
    )
    is_verified = models.BooleanField(
        default=False,
        db_comment="有効化有無",
    )
    created_by = models.ForeignKey(
        "self",
        null=True,
        on_delete=models.SET_NULL,
        related_name="%(class)s_created_by",
        db_comment="作成者",
    )
    updated_by = models.ForeignKey(
        "self",
        null=True,
        on_delete=models.SET_NULL,
        related_name="%(class)s_updated_by",
        db_comment="更新者",
    )

    USERNAME_FIELD = "employee_number"
    REQUIRED_FIELDS = ["email", "username"]

    class Meta:
        ordering = ["employee_number"]
        db_table = "User"
        db_table_comment = "システムユーザ"

    def save(self, *args, **kwargs):
        # 既に登録されているシステム利用者情報の保存処理
        if self.id:
            if not "updated_by" in kwargs:
                self.updated_by = self
            else:
                self.updated_by = kwargs.get("updated_by")
                kwargs.pop("updated_by")
        super(User, self).save(*args, **kwargs)

    def __str__(self):
        return self.username


class Customer(models.Model):
    """お客様"""

    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
        db_comment="ID",
    )
    kana = models.CharField(
        max_length=255,
        db_comment="カナ氏名",
    )
    name = models.CharField(
        max_length=255,
        db_comment="氏名",
    )
    birthday = models.DateField(
        db_comment="誕生日",
    )
    email = models.EmailField(
        db_comment="メールアドレス",
    )
    phone_no = models.CharField(
        max_length=11,
        validators=[RegexValidator(r"^[0-9]{11}$", "11桁の数字を入力してください。")],
        blank=True,
        db_comment="電話番号",
    )
    address = models.OneToOneField(
        "Address",
        on_delete=models.CASCADE,
        related_name="address",
        db_comment="住所のFK",
    )
    created_at = models.DateTimeField(
        auto_now_add=True,
        db_comment="作成日時",
    )
    updated_at = models.DateTimeField(
        auto_now=True,
        db_comment="更新日時",
    )
    created_by = models.ForeignKey(
        User,
        on_delete=models.DO_NOTHING,
        related_name="%(class)s_created_by",
        db_comment="作成者ID",
    )
    updated_by = models.ForeignKey(
        User,
        on_delete=models.DO_NOTHING,
        related_name="%(class)s_updated_by",
        db_comment="更新者ID",
    )

    class Meta:
        db_table = "Customer"


class Address(models.Model):
    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
        db_comment="ID",
    )
    prefecture = models.CharField(
        max_length=255,
        db_comment="都道府県",
    )
    municipalities = models.CharField(
        max_length=255,
        db_comment="市区町村",
    )
    house_no = models.CharField(
        max_length=255,
        db_comment="丁・番地",
    )
    other = models.CharField(
        max_length=255,
        blank=True,
        db_comment="その他(マンション名など)",
    )
    post_no = models.CharField(
        max_length=7,
        validators=[RegexValidator(r"^[0-9]{7}$", "7桁の数字を入力してください。")],
        null=True,
        db_comment="郵便番号",
    )

    class Meta:
        db_table = "Address"


class UploadHistory(models.Model):
    class UploadStatus(models.IntegerChoices):
        WAIT = 1, "処理待ち"
        IN_PROGRESS = 2, "処理中"
        COMPLETE = 3, "正常終了"
        ERROR = 4, "エラー終了"

    id = models.UUIDField(
        primary_key=True,
        default=uuid.uuid4,
        editable=False,
        db_comment="ID",
    )
    csv = models.FileField(
        upload_to="upload_file",
        validators=[
            FileExtensionValidator(
                [
                    "csv",
                ]
            )
        ],
        storage=CustomStorage(),
        db_comment="CSVファイル",
    )
    status = models.IntegerField(
        choices=UploadStatus.choices,
        default=UploadStatus.WAIT,
        db_comment="アップロードステータス",
    )
    created_at = models.DateTimeField(
        auto_now_add=True,
        db_comment="作成日時",
    )
    created_by = models.ForeignKey(
        User,
        on_delete=models.DO_NOTHING,
        related_name="%(class)s_created_by",
        db_comment="作成者ID",
    )

    class Meta:
        db_table = "UploadHistory"

Serializer

CSVをアップロードする用のUploadCSVSerializer
CSV内のデータを登録するリクエストのバリデーションを行うRegisterCSVSerializer
CSV内のデータをInsertする際に使用するAddressSerializer,CustomerSerializer
アップロード履歴一覧用のListUploadHistorySerializer

を作成します
ListUploadHistorySerializerではto_representationを使ったstatusのEnumのLabelの表示とCSVファイルのパスを"/"でsplitして最後の要素のみ表示させるようにします

serializers.py

from rest_framework import serializers

from application.models import Address, Customer, UploadHistory

class AddressSerializer(serializers.ModelSerializer):
    class Meta:
        model = Address
        fields = "__all__"
        read_only_fields = ["id"]


class CustomerSerializer(serializers.ModelSerializer):
    class Meta:
        model = Customer
        fields = [
            "id",
            "name",
            "kana",
            "email",
            "phone_no",
            "birthday",
        ]
        read_only_fields = ["id"]


class ListUploadHistorySerializer(serializers.ModelSerializer):
    class Meta:
        model = UploadHistory
        fields = "__all__"

    def to_representation(self, instance):
        rep = super(ListUploadHistorySerializer, self).to_representation(
            instance
        )
        rep["csv"] = instance.csv.name.split("/")[-1]
        rep["status"] = instance.get_status_display()
        rep["created_by"] = instance.created_by.username
        return rep


class UploadCSVSerializer(serializers.ModelSerializer):
    class Meta:
        model = UploadHistory
        fields = ["id", "csv"]


class RegisterCSVSerializer(serializers.ModelSerializer):
    token = serializers.CharField()

    class Meta:
        model = UploadHistory
        fields = ["id", "token"]

Viewの作成

CSVアップロード用のAPI
CSV内のデータをinsertする用のAPI
履歴一覧表示用のAPI

の3つについて順番に解説します

views.py

import csv

import chardet
from django.conf import settings
from django.db import transaction
from django.http import HttpResponseBadRequest, JsonResponse
from rest_framework import mixins, status
from rest_framework.decorators import action
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet, ModelViewSet

from application.models import Address, Customer, UploadHistory
from application.serializers.customer import (
    AddressSerializer,
    CustomerSerializer,
    ListUploadHistorySerializer,
    RegisterCSVSerializer,
    UploadCSVSerializer,
)
from application.utils.sqs import sqs_send_message


class CustomerViewSet(ModelViewSet):
    queryset = (
        Customer.objects.select_related("address")
        .select_related("created_by")
        .all()
    )

    def get_serializer_class(self):
        match self.action:
            case "upload_csv":
                return UploadCSVSerializer
            case "register_csv":
                return RegisterCSVSerializer

    @action(methods=["post"], detail=False, permission_classes=[IsAuthenticated])
    def upload_csv(self, request):
        """CSVアップロードAPI

        CSVファイルをアップロードする

        Returns:
            JsonResponse
        """
        csv = request.FILES.get("csv")
        encoding = chardet.detect(csv.read())
        csv.seek(0)
        if encoding["encoding"] != "utf-8":
            return JsonResponse(
                data={"msg": "CSVファイルのエンコードをutf-8にしてください"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        serializer = self.get_serializer(
            data={
                "csv": csv,
            }
        )
        serializer.is_valid(raise_exception=True)
        serializer.save(created_by=request.user)
        id = serializer.data["id"]
        sqs_send_message(id)
        return JsonResponse(
            {"msg": "CSVファイルのアップロードに成功しました", "id": id},
            status=status.HTTP_201_CREATED,
        )

    @transaction.atomic
    @action(detail=True, methods=["post"], permission_classes=[AllowAny])
    def register_csv(self, request, pk):
        """Lambdaを使った非同期でお客様情報を登録するAPI

        CSVファイルを読み込み、お客様を非同期で一括で登録する

        Returns:
            - JsonResponse
            - HttpResponseBadRequest
        """
        if request.data.get("token", "") != settings.LAMBDA_TOKEN:
            # 攻撃者へ情報を与えない為にレスポンスにメッセージを入れない
            return HttpResponseBadRequest()
        try:
            entry = UploadHistory.objects.get(pk=pk)
        except UploadHistory.DoesNotExist:
            return JsonResponse(
                data={"msg": "存在しないIDです"},
                status=status.HTTP_404_NOT_FOUND,
            )
        if entry.status != UploadHistory.UploadStatus.WAIT:
            return JsonResponse(
                data={"msg": "処理対象のIDが無効です。"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        try:
            entry.status = UploadHistory.UploadStatus.WAIT
            entry.save()
            decoded_file = entry.csv.read().decode("utf-8").splitlines()
            reader = csv.reader(decoded_file)
            next(reader)
            for row in reader:
                customer_data = {
                    "name": row[0],
                    "kana": row[1],
                    "birthday": row[2],
                    "email": row[3],
                    "phone_no": "0" + row[4],
                }
                address_data = {
                    "prefecture": row[5],
                    "municipalities": row[6],
                    "house_no": row[7],
                    "other": row[8],
                    "post_no": row[9],
                }
                address_serializer = AddressSerializer(data=address_data)
                address_serializer.is_valid(raise_exception=True)
                customer_serializer = CustomerSerializer(data=customer_data)
                customer_serializer.is_valid(raise_exception=True)
                address = Address.objects.create(
                    **address_data,
                )
                Customer.objects.create(
                    **customer_data,
                    address=address,
                    created_by=request.user,
                    updated_by=request.user,
                )
            entry.status = UploadHistory.UploadStatus.COMPLETE
            entry.save()
        except BaseException as e:
            entry.status = UploadHistory.UploadStatus.ERROR
            entry.save()
            return JsonResponse(
                data={"msg": e},
                status=status.HTTP_400_BAD_REQUEST,
            )
        return JsonResponse(
            data={"msg": "CSVファイルの取り込みに成功しました"},
            status=status.HTTP_201_CREATED,
        )


class ListUploadCSVHistoryViewSet(mixins.ListModelMixin, GenericViewSet):
    queryset = UploadHistory.objects.select_related("created_by")
    permission_classes = [IsAuthenticated]
    serializer_class = ListUploadHistorySerializer

    def list(self, request):
        """CSVアップロード履歴

        CSVアップロード履歴を一覧表示する

        Returns:
            Response
        """
        queryset = self.filter_queryset(self.get_queryset())

        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

CSVアップロード用のAPI

まず、リクエスト内のCSVがutf-8の文字コードを使用しているか確認します
その後、CSVファイルなのかバリデーションし、serializer.is_validがtrueの時UploadHistoryを作成し、PKの値をSQSへメッセージとして送信します

    @action(methods=["post"], detail=False, permission_classes=[IsAuthenticated])
    def upload_csv(self, request):
        """CSVアップロードAPI

        CSVファイルをアップロードする

        Returns:
            JsonResponse
        """
        csv = request.FILES.get("csv")
        encoding = chardet.detect(csv.read())
        csv.seek(0)
        if encoding["encoding"] != "utf-8":
            return JsonResponse(
                data={"msg": "CSVファイルのエンコードをutf-8にしてください"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        serializer = self.get_serializer(
            data={
                "csv": csv,
            }
        )
        serializer.is_valid(raise_exception=True)
        serializer.save(created_by=request.user)
        id = serializer.data["id"]
        sqs_send_message(id)
        return JsonResponse(
            {"msg": "CSVファイルのアップロードに成功しました", "id": id},
            status=status.HTTP_201_CREATED,
        )

CSV内のデータをinsertする用のAPI

下記のコードはLambdaで実行するのでpermission_classesをAllowAnyにします
まず、不正アクセス対策のためにtokenの値を確認し、一致しなければ400を返します
次にステータスの値から処理待ち以外だと400を返すようにします
その後、処理中にステータスを変更し、CSV内のデータをinsertします
処理が完了したらステータスをCOMPLETEに変更して終了です

    @transaction.atomic
    @action(detail=True, methods=["post"], permission_classes=[AllowAny])
    def register_csv(self, request, pk):
        """Lambdaを使った非同期でお客様情報を登録するAPI

        CSVファイルを読み込み、お客様を非同期で一括で登録する

        Returns:
            - JsonResponse
            - HttpResponseBadRequest
        """
        if request.data.get("token", "") != settings.LAMBDA_TOKEN:
            # 攻撃者へ情報を与えない為にレスポンスにメッセージを入れない
            return HttpResponseBadRequest()
        try:
            entry = UploadHistory.objects.get(pk=pk)
        except UploadHistory.DoesNotExist:
            return JsonResponse(
                data={"msg": "存在しないIDです"},
                status=status.HTTP_404_NOT_FOUND,
            )
        if entry.status != UploadHistory.UploadStatus.WAIT:
            return JsonResponse(
                data={"msg": "処理対象のIDが無効です。"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        try:
            entry.status = UploadHistory.UploadStatus.IN_PROGRESS
            entry.save()
            decoded_file = entry.csv.read().decode("utf-8").splitlines()
            reader = csv.reader(decoded_file)
            next(reader)
            for row in reader:
                customer_data = {
                    "name": row[0],
                    "kana": row[1],
                    "birthday": row[2],
                    "email": row[3],
                    "phone_no": "0" + row[4],
                }
                address_data = {
                    "prefecture": row[5],
                    "municipalities": row[6],
                    "house_no": row[7],
                    "other": row[8],
                    "post_no": row[9],
                }
                address_serializer = AddressSerializer(data=address_data)
                address_serializer.is_valid(raise_exception=True)
                customer_serializer = CustomerSerializer(data=customer_data)
                customer_serializer.is_valid(raise_exception=True)
                address = Address.objects.create(
                    **address_data,
                )
                Customer.objects.create(
                    **customer_data,
                    address=address,
                    created_by=request.user,
                    updated_by=request.user,
                )
            entry.status = UploadHistory.UploadStatus.COMPLETE
            entry.save()
        except BaseException as e:
            entry.status = UploadHistory.UploadStatus.ERROR
            entry.save()
            return JsonResponse(
                data={"msg": e},
                status=status.HTTP_400_BAD_REQUEST,
            )
        return JsonResponse(
            data={"msg": "CSVファイルの取り込みに成功しました"},
            status=status.HTTP_201_CREATED,
        )

履歴一覧表示用のAPI

表示内容についてはSerializerで制御しているのでrest_frameworkのListModelMixinのコードをそのまま流用します

class ListUploadCSVHistoryViewSet(mixins.ListModelMixin, GenericViewSet):
    queryset = UploadHistory.objects.select_related("created_by")
    permission_classes = [IsAuthenticated]
    serializer_class = ListUploadHistorySerializer

    def list(self, request):
        """CSVアップロード履歴

        CSVアップロード履歴を一覧表示する

        Returns:
            Response
        """
        queryset = self.filter_queryset(self.get_queryset())

        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

実際に検証してみよう！

APIを個別に検証

CSVファイルをアップロードします

以下のようにmsgとidが表示されたら成功です

CSVのimport処理を検証します
以下のように表示されたら成功です

履歴一覧を表示させます
以下のように表示されたら成功です

Localstackを検証

CSVをアップロードするAPIを実行するとDjangoとLocalStackで下記のようなログが出れば成功です

2024-08-12 10:52:37 [12/Aug/2024 10:52:37] "POST /api/customers/upload_csv HTTP/1.1" 201 168 
2024-08-12 10:52:38 [12/Aug/2024 10:52:38] "POST /api/customers/register_csv HTTP/1.1" 201 110

2024-08-12 10:37:38 time="2024-08-12T01:37:38Z" level=warning msg="Cannot list external agents" func=go.amzn.com/lambda/agents.ListExternalAgentPaths file="/home/runner/work/lambda-runtime-init/lambda-runtime-init/lambda/agents/agent.go:24" error="open /opt/extensions: no such file or directory"
2024-08-12 10:37:38 c23e8b17-ec15-4107-8c35-41d0befe6ca8
2024-08-12 10:37:38 [('Date', 'Mon, 12 Aug 2024 01:37:38 GMT'),
2024-08-12 10:37:38  ('Server', 'WSGIServer/0.2 CPython/3.11.9'),
2024-08-12 10:37:38  ('Content-Type', 'application/json'),
2024-08-12 10:37:38  ('Vary', 'Accept, Cookie, Origin'),
2024-08-12 10:37:38  ('Allow', 'POST, OPTIONS'),
2024-08-12 10:37:38  ('djdt-store-id', 'bbaa2ab2a43e447ab2e16f5b1147b4f1'),
2024-08-12 10:37:38  ('Server-Timing',
2024-08-12 10:37:38   'TimerPanel_utime;dur=63.471000000000274;desc="User CPU time", '
2024-08-12 10:37:38   'TimerPanel_stime;dur=1.6119999999999468;desc="System CPU time", '
2024-08-12 10:37:38   'TimerPanel_total;dur=65.08300000000023;desc="Total CPU time", '
2024-08-12 10:37:38   'TimerPanel_total_time;dur=76.9503116607666;desc="Elapsed time", '
2024-08-12 10:37:38   'SQLPanel_sql_time;dur=6.563425064086914;desc="SQL 9 queries", '
2024-08-12 10:37:38   'CachePanel_total_time;dur=0;desc="Cache 0 Calls"'),
2024-08-12 10:37:38  ('X-Frame-Options', 'DENY'),
2024-08-12 10:37:38  ('Content-Length', '110'),
2024-08-12 10:37:38  ('X-Content-Type-Options', 'nosniff'),
2024-08-12 10:37:38  ('Referrer-Policy', 'same-origin'),
2024-08-12 10:37:38  ('Cross-Origin-Opener-Policy', 'same-origin')]
2024-08-12 10:37:38 201
2024-08-12 10:37:38 {'msg': 'CSVファイルの取り込みに成功しました'}

まとめ

今回はLocalStackを使った非同期処理の実装方法を説明するためにAPIの実装をやや簡素なものにしましたが、CSV内のヘッダのバリデーションやエラーが発生した時にエラ〜メッセージを履歴一覧に表示させるなど顧客のニーズに応じて適宜処理を追加しても面白いのではないかと思います
やるべきことがとても多かったですが非同期処理はとても汎用的なものなので一度作成すればたプロジェクトでも横展開できるので学んだことをしっかり復習して同様の実装を任されても再現できるようにしたいです

参考