PythonでcsvファイルからDBにレコードを登録する方法をご紹介します。
条件
- Python 3.7.0
- sqlite3
前提
Djangoで定義した以下のようなモデルが存在するものとします。
# 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（sqlite3）に登録するサンプルです。
import csv
import os.path
from datetime import datetime as dt
from logging import getLogger, StreamHandler, Formatter, DEBUG, FileHandler
import sqlite3
from contextlib import closing
## ログ出力設定
logger = getLogger("気象データ登録")
logger.setLevel(DEBUG)
# コンソール出力設定
stream_handler = StreamHandler()
formatter = Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
# 登録対象csv名
FILE_NAME = 'data.csv'
# DB名
DB_NAME = 'db.sqlite3'
# DBへの追加用SQL
sql_insert = ("insert into weather_data (data_datetime, temperature, humidity, location_id, created_at, updated_at) "
"select * from (select ? as data_datetime, ? as temperature, ? as humidity, ? as location_id, "
"? as created_at, ? as updated_at) as tmp "
"where not exists (select * from weather_data where location_id = ? and data_datetime = ?)")
def regist_data(db):
## ファイル読み込み（CSV形式）
try:
file = open(FILE_NAME, newline='')
except IOError:
logger.warning('対象ファイルが存在しません：' + FILE_NAME)
logger.warning('DB登録は行いません：' + FILE_NAME)
else:
logger.info('=== > Start DB登録 ==')
cursor = db.cursor()
with file:
reader = csv.reader(file)
header = next(reader) # ヘッダーをスキップ
for row in reader:
str_time = [dt.now().strftime('%Y-%m-%d %H:%M:%S')]
add_data = [] # ロケーションID
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
add_data.append(row[0]) # 対象日時（対象レコードがDBに存在するかの確認に使用する）
logger.debug('add_data = ' + str(add_data))
#logger.debug('add_data = ' + sql_insert)
# レコード追加
cursor.execute(sql_insert, add_data)
# コミット
db.commit()
logger.info("=== > End DB登録 ==")
# DB後処理
cursor.close()
### 実処理（main） ###
def main():
logger.info('== batch処理開始 ==')
# DB処理
base_dir = os.path.dirname(os.path.abspath(__file__))
db_path = os.path.join(base_dir, DB_NAME)
logger.info('db_path = ' + db_path)
with closing(sqlite3.connect(db_path)) as db:
regist_data(db)
logger.info('== batch処理終了 ==')
## main関数を実行
if __name__ == '__main__':
main()
大まかな処理の流れは以下の通りです。
- DBに接続
- csvファイル読み込み
- csvファイルから読み込んだデータを1行ずつDBに追加
- DBコミット
バッチ実行結果
登録するデータ(deta.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
バッチ実行前
気象データは空の状態です。
バッチ実行後
db.sqlite3、バッチファイル、データファイル(data.csv)を同じディレクトリに配置して、バッチファイルを実行します。
2018-12-16 15:51:33,952 - 気象データ登録 - INFO - == batch処理開始 == 2018-12-16 15:51:33,952 - 気象データ登録 - INFO - db_path = C:\Users\kmurata\PycharmProjects\graph\db.sqlite3 2018-12-16 15:51:33,953 - 気象データ登録 - INFO - === > Start DB登録 == 2018-12-16 15:51:33,953 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 00:00:00', '5', '10', '1', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '1', '2018-12-16 00:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 01:00:00', '1', '50', '1', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '1', '2018-12-16 01:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 02:00:00', '5', '20', '1', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '1', '2018-12-16 02:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 03:00:00', '10', '70', '1', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '1', '2018-12-16 03:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 04:00:00', '15', '20', '1', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '1', '2018-12-16 04:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 00:00:00', '25', '80', '2', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '2', '2018-12-16 00:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 01:00:00', '12', '20', '2', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '2', '2018-12-16 01:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 02:00:00', '5', '50', '2', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '2', '2018-12-16 02:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 03:00:00', '30', '30', '2', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '2', '2018-12-16 03:00:00'] 2018-12-16 15:51:33,962 - 気象データ登録 - DEBUG - add_data = ['2018-12-16 04:00:00', '15', '70', '2', '2018-12-16 15:51:33', '2018-12-16 15:51:33', '2', '2018-12-16 04:00:00'] 2018-12-16 15:51:33,967 - 気象データ登録 - INFO - === > End DB登録 == 2018-12-16 15:51:33,967 - 気象データ登録 - INFO - == batch処理終了 ==
気象データが追加されています。
参考
サンプルソースをGitHubに公開しています。