Perform loaddata fixture from S3 (from lambda)

django_loaddata_from_s3.py

import sys
from pathlib import Path

from django.core import management
from django.core.management.base import BaseCommand

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

config = Config(
    region_name="ap-northeast-1",
    connect_timeout=3,
    retries={"max_attempts": 5},
)
S3_CLIENT = boto3.client("s3", config=config)

DEFAULT_LAMBDA_DIRECTORY = "/tmp"


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("-s", "--source-key", type=str, required=True, help=f"ロードするS3Fixtureファイルのパス")

    def handle(self, *args, **options):
        source_key: Path = options["source_key"]
        lambda_local_filepath = Path(DEFAULT_LAMBDA_DIRECTORY) / source_key

        self.stdout.write(f"source_key={source_key}")
        self.stdout.write(f"lambda_local_filepath={lambda_local_filepath}")

        self.stdout.write(f"Download from S3 s3://{settings.CSV_DOWNLOAD_BUCKETNAME}/{source_key} to {lambda_local_filepath} ...")
        S3_CLIENT.download_file(settings.CSV_DOWNLOAD_BUCKETNAME, lambda_local_filepath.name, str(lambda_local_filepath))
        self.stdout.write(f"Download from S3 s3://{settings.CSV_DOWNLOAD_BUCKETNAME}/{source_key} to {lambda_local_filepath} ... DONE")

        fixture_name = lambda_local_filepath.stem
        with lambda_local_filepath.open("r") as tmp_stdin:
            sin = sys.stdin
            sys.stdin = tmp_stdin
            management.call_command("loaddata", "-", format="json", verbosity=0)
            sys.stdin = sin