テーブルの定義とデータをファイルに。インポートとエクスポート。

テーブルの定義とデータをファイルに。インポートとエクスポート。

import sqlite3
import datetime
import os
import csv

import time # ダミーデータを作るときのsleep用。

class tabledata_file_converter():
    """テーブルの情報をファイルに書き出す、ファイルからテーブルを復元する
    """
    @staticmethod
    def get_table_definition_from_sqlite_master(db_cursor, table_name):
        # dbmsからテーブル定義の情報を取得する
        sql = f"select sql from sqlite_master where type='table' and name='{table_name}'"
        db_cursor.execute(sql)
        row = db_cursor.fetchone()
        if row is None:
            raise
        return row[0]
        
    @staticmethod
    def build_fullpath_table_def_file(table_name, folder_path=None):
        # テーブル定義を書き込むファイルのパスを作る
        post_fix="_table_def.txt"
        if folder_path is None:
            file_fullpath = os.path.dirname(os.path.abspath(__file__)) + os.sep + table_name + post_fix
        else:
            file_fullpath = folder_path + os.sep + table_name + post_fix
        return file_fullpath

    @staticmethod
    def build_fullpath_table_dat_csv(table_name, folder_path=None):
        # テーブルのデータを書き込むファイルのパスを作る
        post_fix="_table_dat.csv"
        if folder_path is None:
            file_fullpath = os.path.dirname(os.path.abspath(__file__)) + os.sep + table_name + post_fix
        else:
            file_fullpath = folder_path + os.sep + table_name + post_fix
        return file_fullpath

    @staticmethod
    def save_def_file(table_name, table_definition, folder_path=None):
        # テーブル定義をファイルに書き込む
        save_file_fullpath = tabledata_file_converter.build_fullpath_table_def_file(table_name, folder_path)

        with open(save_file_fullpath, mode='w', encoding='utf-8') as f:
            f.write(table_definition)

    @staticmethod
    def read_from_def_file(table_name, folder_path=None):
        # テーブル定義をファイルから読み出す。
        read_file_fullpath = tabledata_file_converter.build_fullpath_table_def_file(table_name, folder_path)

        if os.path.isfile(read_file_fullpath) is False:
            raise
        
        with open(read_file_fullpath, mode='r', encoding='utf-8') as f:
            row = f.readline()
        return row
    
    @staticmethod
    def is_exist_table(db_cursor, tale_name):
        #テーブルの存在確認
        db_cursor.execute(f'SELECT COUNT(*) FROM sqlite_master WHERE TYPE="table" AND NAME="{tale_name}"')
        if db_cursor.fetchone() == (0,): #存在しないとき
            return False
        else:
            return True

    @staticmethod
    def list_to_sql_str(list_data):
        '''
        list_to_sql_str(["hoge", 3, 0.14])
        "'hoge',3,0.14" 
        '''
        y=[]
        for item in list_data:
            if type(item) is str:
                y.append("'"+item+"'")
            else:
                y.append(str(item))
        
        return ','.join(y)

    @staticmethod
    def load_from_csv(db_cursor, table_name, folder_path=None):
        # ファイル(データと定義)からテーブルを復元する

        # テーブルの有無確認　テーブルがあればエラー終了させる
        if __class__.is_exist_table(db_cursor, table_name) is True:
            raise

        # テーブルを定義する
        table_def_sql = __class__.read_from_def_file(table_name)
        db_cursor.execute(table_def_sql)

        read_file_fullpath = __class__.build_fullpath_table_dat_csv(table_name, folder_path)

        f = open(read_file_fullpath,mode='r',encoding='utf-8', newline='')
        csv_reader = csv.reader(f, delimiter=',', quotechar='"', lineterminator='\n')

        for row in  csv_reader:
            insert_sql = f"INSERT INTO {table_name} VALUES({ __class__.list_to_sql_str(row)})"
            db_cursor.execute(insert_sql)
        
    @staticmethod
    def save_csv(db_cursor, table_name, folder_path=None):
        save_file_fullpath=tabledata_file_converter.build_fullpath_table_dat_csv(table_name=table_name, folder_path=folder_path)
        
        f = open(save_file_fullpath,mode='w',encoding='utf-8', newline='')
        csv_writer = csv.writer(f, delimiter=',', quotechar='"', lineterminator='\n')
        
        for row in db_cursor.execute(f'SELECT * FROM {table_name}'):
            csv_writer.writerow(row)

        f.close()


if __name__ == '__main__':
    con = sqlite3.connect(":memory:", check_same_thread=False)
    con.isolation_level = None

    cur = con.cursor()
    cur.execute('PRAGMA temp_store=MEMORY;')
    cur.execute('PRAGMA journal_mode=MEMORY;')

    # Create table
    cur.execute('CREATE TABLE stocks (date text, ts timestamp, trans text, symbol text, qty real, price real, add_col integer)')

    # ダミーデータを作る
    for i in range(100):
        # Insert a row of data
        now = datetime.datetime.now()
        cur.execute("INSERT INTO stocks VALUES ('2006-01-05',?,'BUY','RHAT',100,35.14,?)", (now, str(i)))
        time.sleep(0.01)

    # 新しいほうから3件だけを表示する。
    for row in cur.execute("SELECT * From stocks ORDER BY add_col DESC LIMIT 3"):
        print(row)

    # SQL定義の確認
    for row in cur.execute("select sql from sqlite_master where type='table' and name='stocks'"):
        print(row)

    table_definision = tabledata_file_converter.get_table_definition_from_sqlite_master(db_cursor=cur, table_name='stocks')
    print(table_definision)

    # テーブル定義の書き出し
    tabledata_file_converter.save_def_file(table_name='stocks', table_definition=table_definision)

    # テーブル定義の読み出し
    table_definision2 = tabledata_file_converter.read_from_def_file(table_name='stocks')
    print(table_definision2)

    # テーブルデータの書き出し
    tabledata_file_converter.save_csv(cur, "stocks")

    # テーブルの有無
    print(tabledata_file_converter.is_exist_table(cur,"stocks"))
    print(tabledata_file_converter.is_exist_table(cur, "stocksA"))

    # テーブルデータの復元(stocksA向けのファイルを事前に作っておく)
    tabledata_file_converter.load_from_csv(cur, "stocksA")

    # 復元されたデータの表示。新しいほうから3件だけを表示する。
    for row in cur.execute("SELECT * From stocksA ORDER BY add_col DESC LIMIT 3"):
        print(row)

    # 復元されたデータの表示。件数を表示する
    for row in cur.execute("SELECT COUNT(*) From stocksA ORDER BY add_col DESC LIMIT 3"):
        print(row)

    # SQL定義の確認
    for row in cur.execute("select sql from sqlite_master where type='table'"):
        print(row)