サイトアイコン 知的好奇心

DjangoでcsvファイルをアップロードしてDBにレコードを追加する方法（辞書形式で読み込んで登録）

確認飛行物体

以前の２つの記事を組み合わせて、csvファイルを辞書形式で読み込みDBに登録する方法をご紹介します。

条件

モデル

場所：気象データ = 1:N になるようモデルを作成します。

# models.py
from django.db import models
from django.urls import reverse
from datetime import datetime as dt


class Location(models.Model):
    """場所モデル"""
    class Meta:
        db_table = 'location'

    name = models.CharField(verbose_name='ロケーション名', max_length=255)
    memo = models.CharField(verbose_name='メモ', max_length=255, default='', blank=True)
    author = models.ForeignKey(
        'auth.User',
        on_delete=models.CASCADE,
    )
    created_at = models.DateTimeField(verbose_name='登録日時', auto_now_add=True)
    updated_at = models.DateTimeField(verbose_name='更新日時', auto_now=True)

    def __str__(self):
        return self.name

    @staticmethod
    def get_absolute_url(self):
        return reverse('monitor:index')


class WeatherData(models.Model):
    """気象データモデル"""
    class Meta:
        db_table = 'weather_data'
        unique_together = (('location', 'data_datetime'),)

    location = models.ForeignKey(Location, verbose_name='ロケーション', on_delete=models.PROTECT)
    data_datetime = models.DateTimeField(verbose_name='データ日時', default=dt.strptime('2001-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'))
    temperature = models.FloatField(verbose_name='気温')
    humidity = models.FloatField(verbose_name='湿度')
    created_at = models.DateTimeField(verbose_name='登録日時', auto_now_add=True)
    updated_at = models.DateTimeField(verbose_name='更新日時', auto_now=True)

    def __str__(self):
        return self.location.name + ":" + str(self.data_datetime)

ファイルのアップロード処理

ファイルのアップロード処理はすでに実装済みという前提で話を進めます。

csvデータ登録処理

DB関連処理

以前作成したDB処理用のソース（addCsv.py）を修正します。
コメントアウトしている箇所は修正前の処理です。
今回は、csvを辞書形式で読み込み（csv.DictReader(file)）、カラムを指定してデータを取得してSQLのパラメタに渡します。

# addCsv.py

from django.db import connection
import logging
import csv
from datetime import datetime as dt

logger = logging.getLogger('development')

# DBへの追加用SQL
sql_insert = ("insert into weather_data (data_datetime, temperature, humidity, location_id, created_at, updated_at) "
              "select * from (select %s as data_datetime, %s as temperature, %s as humidity, %s as location_id, "
              "%s as created_at, %s as updated_at) as tmp "
              "where not exists (select * from weather_data where location_id = %s and data_datetime = %s)")


def regist_data(cursor, file_path):
    # ファイル読み込み（CSV形式）
    try:
        file = open(file_path, newline='')
    except IOError:
        logger.warning('対象ファイルが存在しません：' + file_path)
        logger.warning('DB登録は行いません：' + file_path)
    else:
        logger.info('=== > Start DB登録 ==')
        with file:
            # reader = csv.reader(file)
            # header = next(reader)  # ヘッダーをスキップ
            reader = csv.DictReader(file)

            for row in reader:
                str_time = [dt.now().strftime('%Y-%m-%d %H:%M:%S')]
                add_data = []
                add_data.append(row.get('data_datetime'))  # csvから辞書形式で読み取った情報
                add_data.append(row.get('temperature'))  # csvから辞書形式で読み取った情報
                add_data.append(row.get('humidity'))  # csvから辞書形式で読み取った情報
                add_data.append(row.get('location_id'))  # csvから辞書形式で読み取った情報
                add_data.extend(str_time)  # created_at
                add_data.extend(str_time)  # updated_at
                add_data.append(row.get('location_id'))  # ロケーションID（対象レコードがDBに存在するかの確認に使用する）
                add_data.append(row.get('data_datetime'))  # 対象日時（対象レコードがDBに存在するかの確認に使用する）
                logger.debug('add_data = ' + str(add_data))

                # レコード追加
                cursor.execute(sql_insert, add_data)


            # for row in reader:
            #     str_time = [dt.now().strftime('%Y-%m-%d %H:%M:%S')]
            #     add_data = []
            #     add_data.extend(row)  # csvから読み取った情報
            #     add_data.extend(str_time)  # created_at
            #     add_data.extend(str_time)  # updated_at
            #     add_data.append(row[3])  # ロケーションID（対象レコードがDBに存在するかの確認に使用する）
            #     add_data.append(row[0])  # 対象日時（対象レコードがDBに存在するかの確認に使用する）
            #     logger.debug('add_data = ' + str(add_data))
            #
            #     # レコード追加
            #     cursor.execute(sql_insert, add_data)

            logger.info("=== > End DB登録 ==")


# csvファイルのデータをDBに追加する。
def insert_csv_data(file_path):
    logger.info('== csvデータ登録処理開始 ==')

    with connection.cursor() as cursor:
        regist_data(cursor, file_path)

    logger.info('== csvデータ登録処理終了 ==')

実行結果

前提

以下のようなレコード状態であるものとします。

レコード状態が適切でないと外部キー制約で追加時にエラーになる場合があります。
例えば今回のモデルでは、存在しないlocation_idのレコードをweather_dataテーブルに追加しようとすると外部キー制約エラーになります。

auth_user

管理者（admin）id=1が存在する。

location

管理者（admin）に紐づくレコードが２つ（id=1,2）が存在する。

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

data_datetime,temperature,humidity,location_id
2018-12-16 00:00:00,5,10,1
2018-12-16 01:00:00,1,50,1
2018-12-16 02:00:00,5,20,1
2018-12-16 03:00:00,10,70,1
2018-12-16 04:00:00,15,20,1
2018-12-16 00:00:00,25,80,2
2018-12-16 01:00:00,12,20,2
2018-12-16 02:00:00,5,50,2
2018-12-16 03:00:00,30,30,2
2018-12-16 04:00:00,15,70,2

アップロード前

weather_dataテーブルにレコードは存在しません。

アップロード後

weather_dataテーブルにレコードが追加されています。

以上で、csvファイルを辞書形式で読み込みDBに登録する方法のご紹介は終わりです。

モバイルバージョンを終了