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

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

条件

  • Django 2.1.4
  • Python 3.7.0

モデル

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

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# 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)
# 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)
# 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のパラメタに渡します。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# 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データ登録処理終了 ==')
# 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データ登録処理終了 ==')
# 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)に紐づくレコードが2つ(id=1,2)が存在する。

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

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
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
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
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に登録する方法のご紹介は終わりです。

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です