環境
Pythonバージョン：3.7.8
開発環境：Mac
※Windowsであれば下記よりPythonのインストーラーをダウンロードして環境構築する
Windows版Pythonのインストール - python.jp
※インストールが必要なライブラリ
* pandas
* psycopg2
<インストールの参考>
WindowsでPython3, numpy, pandas, matplotlibなどインストール - Qiita
目標
AccessのデータをPostgreSQLにデータ移行する
問題点
- カラム内で改行している。
テキストエディターの置換で正規表現を使って改行コードを「LF(\n)」に変換することもできるが、
Pythonを使って業務ツール作った方が、DB接続もできるので便利。
↓↓↓
[解決策] CSVでデータをエクスポートするとCSVのフォーマットが崩れるので、 JSONでエクスポートする。
- カラム数が強烈に多すぎて１万件のinsertですら固まる。
↓↓↓
[解決策] PostgreSQLの「COPY」コマンドを使う。
その他実現したいこと
- PostgreSQLでbool型はchar(1)にしたい。
準備
実装したコード
CSV編集
import glob
import pathlib
import json
import pandas as pd
from collections import OrderedDict
from distutils.util import strtobool
# 【CSVファイル編集変換ツール】
# <環境>
# * 本プログラムのPythonのバージョン：3.7.8
# * WindowsのPython環境構築
# https://www.python.jp/install/windows/install.html
# <実行前準備>
# * DBアクセスツールからエクスポートするときの注意点
# * JSONを選択する
# * ファイル名はインポート先のテーブル名を指定する
# * エクスポートしたファイルは本プログラムと同階層の「json」ディレクトリに配置する
# <実行方法>
# * コマンドプロンプトで本プログラムを格納しているパスに移動する
# $ cd [配置したパス]/python_tool
# * プログラムを実行する。
# $ python edit_files.py
# ★JSONで取り込む
# 1. [OK!]JSONデータを辞書型オブジェクトに変換する
# https://qiita.com/ndj/items/c54a057a014b1977c538
def edit_file(file_name):
d = {}
# 下記encodingの指定はWindowsのみ(Macは不要)
with open("./json/" + file_name, mode="r", encoding="utf-8") as f:
# json.loadは列順が担保されないので下記引数が必要
# https://note.nkmk.me/python-json-load-dump/
d = json.load(f, object_pairs_hook=OrderedDict)
# key = [k for k in d.keys()][0]
# print(key)
list_d = d[[k for k in d.keys()][0]]
# データフレーム型で列順序保持する解決方法1. 辞書型に変換後に列順を保存
# https://www.javadrive.jp/python/dictionary/index8.html#section1
keys = list(list_d[0].keys())
# 2. 下記２種類の三項演算子との組み合わせのリスト内包表記(１行でループ記述できる)でデータを編集する。
# https://note.nkmk.me/python-list-comprehension/
esc_dic_list = []
for dic in list_d:
for t in dic.items():
# a. カラム内の改行コードを「\\n」に変換する...JSONで出力すると「\n」になっているが、
# このままCSVに出力すると改行してしまうので「\\n」とエスケープする必要がある。
if type(t[1]) is str:
dic[t[0]] = t[1].replace("\r", "")
dic[t[0]] = dic[t[0]].replace("\n", "\\n")
dic[t[0]] = dic[t[0]].rstrip(" ")
# bool型の「true」「false」を「'1'」「'0'」に変換する
elif type(t[1]) is bool:
dic[t[0]] = '1' if t[1] else '0'
# 後ほど処理するデータフレームがint型(整数)を自動で小数桁を付加してしまうので、
# int型はstr型に変換する
# (CSVに出力した時、引用符が付かないのでDBのCOPYコマンドを使用する際数値として扱ってもらえる)
elif type(t[1]) is int:
dic[t[0]] = str(t[1])
# print(list_d)
# 3. 辞書型オブジェクトをCSVデータに変換する。
# 辞書リストをデータフレーム型に変換
# https://atsashimipy.hatenablog.com/entry/2019/05/16/150303
# print(pd.options.display.precision)
# https://note.nkmk.me/python-pandas-json-normalize/
# df = pd.io.json.json_normalize(list_d)
df = pd.json_normalize(list_d)
# print("辞書型をデータフレーム型に変換")
# 解決方法2.列順を指定する
# https://note.nkmk.me/python-pandas-reindex/
df_reindex = df.reindex(columns=keys)
# print(df_reindex)
# データフレーム型をCSVに変換
# ※ヘッダー とindexなしにする
# https://note.nkmk.me/python-pandas-to-csv/
# print('***引数のファイル名を編集***')
print(file_name.replace('.json', '.csv'))
df_reindex.to_csv('./csv/' + file_name.replace('.json', '.csv'), header=False, index=False)
# ★ファイル名のみ取得
# print('**************')
# print('ファイル名のみ取得：')
files = pathlib.Path('./json/').glob('*.json')
for file in files:
edit_file(file.name)
# ******【参考】*********
# ★CSVファイルだと１カラム内に改行が入っているとそのまま改行してしまい使えなかった。
# ->JSONファイルでエクスポートすると改行は「\n」で表現されていたので、JSONでのエクスポートを採用した。
# ・行をすべて読み込む
# https://qiita.com/Cesaroshun/items/b331844a54d3618c4c3a
# print('**************')
# print('tables_of_various_types.csvの中身：')
# with open('./csv/tables_of_various_types.csv', mode='r', encoding='utf-8') as f:
# i = 0
# for line in f:
# i = i + 1
# print(i)
# print('行目')
# print(line)
# ★データフレーム型は列の順序を指定しないとアルファベット順になる
# ★path+ファイル名取得
# print('**************')
# print('path+ファイル名取得：')
# files = glob.glob('./json/*')
# for file in files:
# edit_file(file)
PostgreSQLにCSVファイルをCOPYコマンドでインポート
import glob
import pathlib
import psycopg2
from psycopg2.extras import DictCursor
# DB接続
def get_connection():
return psycopg2.connect("dbname='[DB名]' host='[host名またはhostのIPアドレス]' user='[DBのuser]' password='[DBのpassword]'")
# DBインポート処理
def db_import(file_name):
print(file_name)
table_name = file_name.replace('.csv','')
print(table_name)
# 該当テーブルの登録前件数表示
with get_connection() as conn:
with conn.cursor(cursor_factory=DictCursor) as cur:
cur.execute('SELECT COUNT(*) AS count FROM ' + table_name)
row = cur.fetchone()
print('登録前の件数：')
print(row)
# インポートするファイルの件数表示
print('ファイルの件数：')
# リスト内包表記
# https://note.nkmk.me/python-list-comprehension/
print(sum([1 for _ in open('./csv/' + file_name, encoding="utf-8")]))
col_names = []
with get_connection() as conn:
with conn.cursor(cursor_factory=DictCursor) as cur:
# cur.execute("SELECT column_name from information_schema.columns where table_name = 'tables_of_various_types' ORDER by ordinal_position")
cur.execute('SELECT * FROM ' + table_name)
col_names = [col.name for col in cur.description]
# id列を削除(現在serial型にしているから除外)
col_names.pop(0)
#COPYの実行
with get_connection() as conn:
with conn.cursor(cursor_factory=DictCursor) as cur:
with open('./csv/' + file_name, mode='r', encoding='utf-8') as f:
print(col_names)
cur.copy_from( f , table_name, sep=',', null='',columns=(col_names))
# sep='デリミッタ文字種：例ではTAB記号'
# null='ヌル文字種：例ではNULL'...本プログラムでは''にした。
# columns=('') 入力対象のカラム名を順番に列挙、全カラムに投入を可能な場合は省略可能
with get_connection() as conn:
with conn.cursor(cursor_factory=DictCursor) as cur:
cur.execute('SELECT COUNT(*) AS count FROM ' + table_name)
row = cur.fetchone()
print('登録後の件数：')
print(row)
print("OK!")
# ★CSVファイルある分だけインポート処理
files = pathlib.Path('./csv/').glob('*.csv')
for file in files:
db_import(file.name)
