プログラミングと株式投資のブログ

プログラミングで株式投資に役立つ何かをやってます

pythonでsqlite3を使用する際のメモ

Python

ドキュメント

　

インストール

sqlite3はpython3.12では標準で組み込まれているためインストール不要です。
　

sqlite3の使い方

基本的な使い方

import sqlite3

conn = sqlite3.connect(db_path_str)
cur = conn.cursor()

cur.execute(sql_do_something)
conn.commit()

cur.close()
conn.close()

　

カラム名でアクセス可能にする

conn = sqlite3.connect(db_path_str)
self._conn.row_factory = sqlite3.Row

cur = conn.cursor()
cur.execute(sql_get_row)

row = fetchone()
print(row['culumn_name'])

　

SQLの実行はパラメタ化すること

doc_id_strの内容によっては意図しない処理を実行してしまう。例えばdoc_id_strが "0 OR TRUE"のような形だとすべてのレコードを取得されてしまう。SQLの実行の際はSQL文とそのパラメタを分離して渡してやることでこの問題を回避できる。

#BADケース
cur.execute(f"SELSECT * FROM edinet_document WHERE doc_id = {doc_id_str}"

#GOODケース
cur.execute('SELECT * FROM edinet_document WHERE doc_id=?',(doc_id_str,) )

　

DBにテーブルを作成する

関数名はensure_tablesくらいがよさそう。make_tableやinit_dbでは処理内容と名前が合致しない。

CREATE_EDINET_DOC_SQL = '''
CREATE TABLE IF NOT EXISTS edinet_document(
    seq_number TEXT,
    doc_id TEXT PRIMARY KEY,
)
'''

cur.execute(CREATE_EDINET_DOC_SQL)

　

テーブルから特定の行を検索する

RECORD_SELECT_SQL="""
SELECT * FROM edinet_document WHERE doc_id = ?
"""

data_tuple = (doc_id_str, )

#単一
cur.execute(RECORD_SELECT_SQL, data_tuple )
row = cur.fetchone()
print(row)

#複数
cur.execute(RECORD_SELECT_SQL, data_tuple )
rows = cur.fetchall()

for row in rows :
    print(row)

　

DBに特定のテーブルが存在するか判定する

SQL="""
SELECT name FROM sqlite_master 
WHERE type='table' AND name='edinet_document'
"""

cur.execute("SQL")

if cur.fetchone() is not None :
    print('table is exists')
else :
    print('table is not exists')

　

レコードのインサート処理例

重複エラーは呼び出し元でどう扱うか判断させるために別でハンドリングする。その他のエラーとしてはsqlite3.OperationalErrorなどがあるが(DBロック、ファイルI/O)、シングルスレッド・シングルプロセス、ローカルDBアクセスを想定しているため特に何もしない。リモートDBアクセスの場合、通信状況によってエラーが発生する可能性があるためリトライ処理を入れ込むのがよい。
ロールバック失敗時の挙動も検討が必要。ここではロールバック失敗を呼び出し元が検出した場合、処理を直ちに終了し、DBの健全性確認もしくは復旧を行う想定で実装している。システムの稼働を継続する必要がある場合は失敗を保存(通知)し、DB接続のやり直しやDBの簡易な健全性確認を行った上で処理継続することになる。

dao.py

#レコードインサートのSQLとデータ
INSERT_ALREADY_READ_JSON_SQL = '''
INSERT INTO already_read_json (
    json_name
) VALUES(?)'''
values_tuple('test', )

#DB接続
conn = sqlite3.connection('test.db')

#インサート処理本体
cur = None
try :
    cur = conn.cursor()
    cur.execute(INSERT_ALREADY_READ_JSON_SQL , values_tuple)
    conn.commit()
#重複エラー
except sqlite.IntegrityError :
    logger.warning('何かしらのメッセージ')
    #失敗時は必ずロールバックする
    try :
        conn.rollback()
    #ロールバックの失敗もケアする
    except Exception rb_e :
        logger.critical('ロールバックに失敗しました', exc_info=rb_e)
        raise DataBaseRollbackError() from rb_e
    raise
#その他エラー
except Exception :
    logger.exception('何かしらのメッセージ')
    try :
        conn.rollback()
    except Exception rb_e :
        logger.critical('ロールバックに失敗しました', exc_info=rb_e)
        raise DataBaseRollbackError() from rb_e
    raise
finally :
    if cur is not None :
        cur.close()

conn.close()

　
exceptions.py

class DataBaseError(Exception):
    pass
class DataBaseNotConnectedError(DataBaseError):
    pass
class DataBaseRollbackError(DataBaseError):
    pass

　

インサートデータ生成用メソッド

テーブルのカラムやレコードは変わることが多いためレコードからインサート用データを生成するメソッドを用意しておくと便利だし、変化があったときの修正が簡単だ。

@staticmethod
def _map_record_to_tuple(record) :
    return (
        record.get_culumn1(),
        (中略)
        record.get_culumnN(),
    )