Sqlite3接続チェック方法

Question

Sqlite3で接続しデータベースを操作する際、
一度接続後に、ハードディスクの切断などでファイルの接続がなくなり、
その後復帰した場合、プログラムを再起動せずにSqlite3を再接続したいのですが、
このような場合どのようにするのがセオリーですか？

データベースのパスの存在でチェックするしか思い浮かばず。

import sqlite3

dbname = 'TEST.db'
conn = sqlite3.connect(dbname)
cur = conn.cursor()

#ハードディスクの接続が一度切れる

#ここで接続チェックと再接続

cur.execute(
    'CREATE TABLE persons(id INTEGER PRIMARY KEY AUTOINCREMENT,
     name STRING)')

conn.commit()
conn.close()

Answer 1

すごい状況ですが、普通なら DB接続( conn = sqlite3.connect(dbname) )からやり直すと思います。
例えば、 tenacity というライブラリがあって、
(参考: https://ohke.hateblo.jp/entry/2020/11/21/230000)
こういうのを使って、接続からSQL実行まで何度かretryするという方法はあるかと。

from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt

# IOErrorの場合、30秒間隔で 3回まで retryする
@retry(wait=wait_fixed(30), retry=retry_if_exception_type(IOError), stop=stop_after_attempt(3))
def exec_query(dbname, sql):
  conn = sqlite3.connect(dbname)
  cur = conn.cursor()
  cur.execute(sql)
  conn.commit()
  conn.close()

exec_query('TEST.db', 'CREATE TABLE persons(id INTEGER PRIMARY KEY AUTOINCREMENT,
     name STRING)')

• IOError のときにRetryすれば良いのかはちょっと不明(HDDがなくなるとどういうエラーになるか次第...)
• ループ回数が多い中でこれを使うと、毎回接続したりしているので overhead がひどいかもしれない
  • もうちょっと賢く、エラーになった場合だけ再接続するような書き方も考えられます

ご参考まで。

Comments

[退会済みユーザー]

ありがとうございます！
すごく参考になります。