■pythonからPostgreSQLへ接続し、cursorまでできている。SQLでINSERTする前にPostgreSQLとの接続が継続できているか確認する方法があるか？
●AI回答
PythonからPostgreSQLへの接続（一般的に広く使われているライブラリである psycopg2 や psycopg を想定します）において、INSERT前に接続が生きているか確認するには、「コネクションオブジェクトのステータス（属性）をチェックする方法」と「軽量なダミーSQLを実行する方法」の2つがあります。
利用しているライブラリのバージョンや、どの程度厳密にチェックしたいかによって使い分けてください。
方法1：connection.closed 属性を確認する（最も軽量）
プログラム側で明示的にクローズしたか、あるいはライブラリ側で検知できている切断状態を瞬時に判定できます。サーバーへの問い合わせ（ネットワーク通信）が発生しないため非常に高速です。
python
# connection.closed が 0 なら接続中、0 以外なら切断されている
if connection.closed == 0:
# 接続が継続しているので INSERT を実行
cur.execute("INSERT INTO ...")
else:
print("接続が切断されています。再接続が必要です。")
コードは注意してご使用ください。
• 注意点: ネットワークの物理的な断線や、PostgreSQLサーバーが急に強制終了した直後など、「まだPython側が気づいていない予期せぬ切断」は検知できない場合があります。
方法2：ダミーSQL SELECT 1; を実行する（最も確実）
実際に生存確認（Ping）用の超軽量SQLを一度実行してみる方法です。ネットワークが生きているかを確実に担保できます。
python
try:
# 最も負荷の低いダミーSQLを実行して生存確認
cur.execute("SELECT 1;")
cur.fetchone()
# 成功したら INSERT を実行
cur.execute("INSERT INTO ...")
except (psycopg2.InterfaceError, psycopg2.OperationalError) as e:
print(f"接続が切れています。再接続を試みます: {e}")
# ここで再接続処理を行う
コードは注意してご使用ください。
• メリット: 実際にサーバーと通信するため、完全な生存確認になります。
• 注意点: わずかですが通信のオーバーヘッドが発生します。
おすすめの設計：try-except でエラーをキャッチする
事前にチェックするよりも、「INSERT自体を try-except で囲み、切断エラーが起きたら再接続してリトライする」 という実装（EAFP原則）が、データベースプログラミングでは最も一般的かつ安全です。
事前に生存確認をしても、確認した直後、INSERTを実行するミリ秒の間に接続が切れる可能性（Time-of-check to time-of-use） を排除できないためです。
python
import psycopg2
def execute_insert(connection, cur, sql, data):
try:
cur.execute(sql, data)
connection.commit()
except (psycopg2.InterfaceError, psycopg2.OperationalError):
print("接続が切断されていたため、再接続します...")
# ここで connection と cursor を再生成する処理
connection, cur = reconnect()
# 再度実行
cur.execute(sql, data)
connection.commit()
コードは注意してご使用ください。
より具体的な実装に落とし込むために、もしよろしければ以下について教えてください。
• 使用しているライブラリは psycopg2 ですか？ それとも新しい psycopg (v3) ですか？
• 今回のスクリプトは、常時起動して断続的にINSERTを行うバッチ（Botやスクレイピングなど） ですか？
■psycopg (v3)を使う。プログラムはMQTTのサブスクライバである。プログラムでMQTTデータを受取、データをPostgreSQLに書き込む。エラーが出たらSQLiteにデータを書き込む。SQLiteはサブスクライバと同じマシン上にある。
●AI回答
psycopg (v3) と MQTT サブスクライバを組み合わせる場合、ネットワーク瞬断や DB サーバーの一時停止に備え、「try-except でエラーをキャッチし、即座にローカルの SQLite へ退避させる」 設計が最も堅牢で確実です。
psycopg (v3) では、接続状態の確認に connection.closed 属性や、より詳細な状態を返す connection.info.status が使えます。しかし、MQTT のように非同期でデータが飛んでくる環境では、事前チェックをすり抜けて INSERT 時にエラーになるケースがあるため、以下のように実装するのが最適です。
堅牢な実装コード例
MQTT のメッセージ受信コールバック（on_message）内で、PostgreSQL への書き込みと、失敗時の SQLite へのバックアップを制御する構造です。
python
import psycopg
import sqlite3
import paho.mqtt.client as mqtt
# PostgreSQLの接続設定
PG_CONNINFO = "dbname=test user=postgres password=secret host=localhost"
# ローカルSQLiteのファイルパス
SQLITE_DB_PATH = "local_backup.db"
# SQLiteの初期化（テーブルがなければ作成）
def init_sqlite():
with sqlite3.connect(SQLITE_DB_PATH) as s_conn:
s_conn.execute("""
CREATE TABLE IF NOT EXISTS mqtt_backup (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT,
payload TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
s_conn.commit()
# PostgreSQLへの接続・再接続を試みる関数
def get_pg_connection():
try:
return psycopg.connect(PG_CONNINFO)
except psycopg.OperationalError:
return None
# グローバルまたはクラス変数でコネクションを管理
pg_conn = get_pg_connection()
init_sqlite()
def on_message(client, userdata, msg):
global pg_conn
payload_str = msg.payload.decode('utf-8')
# 1. PostgreSQLの接続状態チェック（事前確認）
# ※ pg_conn が None、またはすでに閉じられている場合
if pg_conn is None or pg_conn.closed:
print("PostgreSQLに接続されていないため、再接続を試みます...")
pg_conn = get_pg_connection()
# 2. PostgreSQLへのインサートを試行
if pg_conn and not pg_conn.closed:
try:
# psycopg v3 では connection.cursor() をコンテキストマネージャで使うのが推奨
with pg_conn.cursor() as cur:
cur.execute(
"INSERT INTO mqtt_logs (topic, payload) VALUES (%s, %s)",
(msg.topic, payload_str)
)
pg_conn.commit()
print(f"[PG] 保存成功: {msg.topic}")
return # 成功したらここで終了
except (psycopg.OperationalError, psycopg.InterfaceError) as e:
# 事前チェックをすり抜けた、実行直前の切断エラーなどをキャッチ
print(f"[PG] 送信中に接続エラーが発生しました: {e}")
# エラーが起きたコネクションは明示的に閉じる
try:
pg_conn.close()
except:
pass
pg_conn = None # 次回メッセージ受信時に再接続を促す
except Exception as e:
# SQL構文エラーなど、接続以外のエラー
print(f"[PG] その他のエラー: {e}")
# 3. PostgreSQLへの書き込みが失敗、または接続できなかった場合は SQLite へ退避
print(f"[SQLite] PostgreSQLが利用不可のため、ローカルに退避します: {msg.topic}")
try:
with sqlite3.connect(SQLITE_DB_PATH) as s_conn:
s_conn.execute(
"INSERT INTO mqtt_backup (topic, payload) VALUES (?, ?)",
(msg.topic, payload_str)
)
s_conn.commit()
except Exception as se:
print(f"[Fatal] SQLiteへの書き込みにも失敗しました: {se}")
# MQTTクライアントの設定（Paho MQTT v2準拠の例）
# ... (MQTTの接続設定やループ起動の処理) ...
コードは注意してご使用ください。
psycopg (v3) におけるポイント
1. connection.closed の活用
コード内で pg_conn.closed を確認しています。これは psycopg v3 でも有効で、True（閉じている）か False（開いている）の真偽値を返します。
2. connection.info.status（より高度な確認）
もし「トランザクションが異常な状態のまま固まっていないか」まで厳密に調べたい場合は、pg_conn.info.status を確認します。これが psycopg.pq.ConnStatus.OK であれば、コネクション自体は正常です。
3. エラーハンドリングの網羅
ネットワーク断絶やサーバー停止時は psycopg.OperationalError または psycopg.InterfaceError が発生します。これらを特定してキャッチすることで、DB起因のトラブル時に確実に SQLite 側へ処理を流せます。
この構成により、PostgreSQLが落ちている間はSQLiteにデータが溜まり続けます。
• 今後、「PostgreSQLが復旧した時に、SQLiteに溜まったデータを自動でPostgreSQLへ再転送（同期）する仕組み」 も必要になりますか？
• 必要であれば、バックグラウンドのスレッドなどで安全に同期するアプローチを提案できますが、いかがでしょうか？
■PostgreSQLが復旧した時に、SQLiteに溜まったデータを自動でPostgreSQLへ再転送（同期）する仕組み」 も必要。サブスクライバはLINUX上で常時動作。
●AI回答
Linux上で常時動作するMQTTサブスクライバにおいて、メインの受信処理（リアルタイム書き込み）を邪魔せずに、SQLiteに溜まったバックアップデータをPostgreSQLへ自動同期するには、「別スレッド（バックグラウンドタスク）で定期的に復旧監視と同期を行う仕組み」が最もスマートで安全です。
MQTTのメッセージ受信コールバック（on_message）と同じスレッドで同期処理（SELECTやINSERTのループ）をしてしまうと、同期中に次のMQTTメッセージを取りこぼす（タイムアウトやパケットロス）原因になります。
以下に、Paho MQTT、psycopg (v3)、sqlite3、そしてバックグラウンド同期スレッドを組み合わせた、Linuxシステム向けの完全な実装コードを示します。
自動同期機能付き MQTTサブスクライバ実装コード
python
import time
import threading
import sqlite3
import psycopg
import paho.mqtt.client as mqtt
# --- 設定 ---
PG_CONNINFO = "dbname=test user=postgres password=secret host=localhost"
SQLITE_DB_PATH = "local_backup.db"
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
MQTT_TOPIC = "device/data"
# スレッド間で共有するPostgreSQLコネクション用のロック
# (psycopg3の接続・切断操作を安全に行うため)
pg_lock = threading.Lock()
pg_conn = None
def get_pg_connection():
"""PostgreSQLへの接続を試みる関数"""
try:
return psycopg.connect(PG_CONNINFO)
except psycopg.OperationalError:
return None
def init_sqlite():
"""ローカルSQLiteの初期化"""
with sqlite3.connect(SQLITE_DB_PATH) as s_conn:
s_conn.execute("""
CREATE TABLE IF NOT EXISTS mqtt_backup (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT,
payload TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
s_conn.commit()
# --- バックグラウンド同期タスク ---
def sync_worker():
"""PostgreSQLの復旧を監視し、SQLiteのデータを再転送するスレッド"""
global pg_conn
print("[Sync] 同期監視スレッドが起動しました。")
while True:
# 10秒ごとに復旧チェックと同期を試みる
time.sleep(10)
with pg_lock:
# PostgreSQLが未接続なら接続を試みる
if pg_conn is None or pg_conn.closed:
pg_conn = get_pg_connection()
if pg_conn is None:
# まだ復旧していない場合は次のサイクルへ
continue
else:
print("[Sync] PostgreSQLへの接続が確認（または復旧）されました。")
# SQLiteに未同期データがあるか確認
try:
with sqlite3.connect(SQLITE_DB_PATH) as s_conn:
s_conn.row_factory = sqlite3.Row
rows = s_conn.execute("SELECT id, topic, payload FROM mqtt_backup ORDER BY id ASC LIMIT 100").fetchall()
if not rows:
continue # データが空なら何もしない
print(f"[Sync] SQLiteに {len(rows)} 件の未同期データを見つけました。転送を開始します...")
# PostgreSQLへ一括挿入（バルクインサート）を試みる
with pg_conn.cursor() as pg_cur:
success_ids = []
for row in rows:
try:
pg_cur.execute(
"INSERT INTO mqtt_logs (topic, payload) VALUES (%s, %s)",
(row["topic"], row["payload"])
)
success_ids.append(row["id"])
except (psycopg.OperationalError, psycopg.InterfaceError) as pg_err:
# 同期中にPostgreSQLが再度落ちた場合
print(f"[Sync] 同期中に再切断されました: {pg_err}")
raise pg_err # 上位のexceptへ抜けてコネクションをクローズさせる
except Exception as e:
# SQLエラー（データ不良など）の場合は、その行をスキップするかログに残す
print(f"[Sync] 特定データのインサートに失敗（スキップします）: {e}")
# ここでは一旦成功扱いにしてSQLiteから消すか、エラーテーブルに移すのが実務上安全
success_ids.append(row["id"])
# コミット
pg_conn.commit()
# PostgreSQLへの書き込みが成功したレコードをSQLiteから削除
if success_ids:
s_conn.execute(
f"DELETE FROM mqtt_backup WHERE id IN ({','.join(['?']*len(success_ids))})",
success_ids
)
s_conn.commit()
print(f"[Sync] {len(success_ids)} 件のデータをPostgreSQLに同期し、ローカルから削除しました。")
except (psycopg.OperationalError, psycopg.InterfaceError):
# 同期中に落ちた場合のハンドリング
try:
pg_conn.close()
except:
pass
pg_conn = None
except Exception as e:
print(f"[Sync] 同期処理中に予期せぬエラー: {e}")
# --- MQTT メッセージ受信コールバック ---
def on_message(client, userdata, msg):
global pg_conn
payload_str = msg.payload.decode('utf-8')
with pg_lock:
# 1. PostgreSQLの生存チェックと書き込み試行
if pg_conn and not pg_conn.closed:
try:
with pg_conn.cursor() as cur:
cur.execute(
"INSERT INTO mqtt_logs (topic, payload) VALUES (%s, %s)",
(msg.topic, payload_str)
)
pg_conn.commit()
print(f"[PG] リアルタイム保存成功: {msg.topic}")
return # 成功したらここで終了
except (psycopg.OperationalError, psycopg.InterfaceError) as e:
print(f"[PG] 送信中に切断を検知しました: {e}")
try:
pg_conn.close()
except:
pass
pg_conn = None # スレッド側、または次回の受信で再接続させる
# 2. PostgreSQLが使えない場合はSQLiteへ即座に退避
print(f"[SQLite] PostgreSQL利用不可のためローカルに退避: {msg.topic}")
try:
with sqlite3.connect(SQLITE_DB_PATH) as s_conn:
s_conn.execute(
"INSERT INTO mqtt_backup (topic, payload) VALUES (?, ?)",
(msg.topic, payload_str)
)
s_conn.commit()
except Exception as se:
print(f"[Fatal] SQLiteへの書き込みも失敗: {se}")
# --- メイン処理 ---
def main():
global pg_conn
init_sqlite()
# 初回接続試行
pg_conn = get_pg_connection()
# 同期用のバックグラウンドスレッドを起動
# daemon=True にすることでメインプログラム終了時に一緒に終了させる
checker_thread = threading.Thread(target=sync_worker, daemon=True)
checker_thread.start()
# MQTTクライアントの起動 (Paho MQTT v2系の記述例)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_message = on_message
print("MQTT Broker に接続中...")
client.connect(MQTT_BROKER, MQTT_PORT, 60)
client.subscribe(MQTT_TOPIC)
# ループを開始（常時待機）
client.loop_forever()
if __name__ == "__main__":
main()
コードは注意してご使用ください。
この設計のメリットとLinux運用でのポイント
1. threading.Lock() による安全性の確保
メインのMQTT受信スレッドと、バックグラウンドの同期スレッドが、同時に同じ PostgreSQL のコネクションを操作して衝突（クラッシュ）しないように with pg_lock: で排他制御を行っています。
2. LIMIT 100 による段階的な同期
もしPostgreSQLが数日間落ちていてSQLiteに大量のデータが溜まった場合、一気に全件を転送しようとするとメモリやネットワークを圧迫します。コード例では LIMIT 100 を設定し、1回につき最大100件ずつ、小分けにして安全に同期するようにしています。
3. Linuxでの常時動作（デーモン化）
Linux上でこのスクリプトを「サーバー再起動時にも自動起動させ、落ちたら自動復旧する」ように運用するには、systemd のサービスとして登録するのが最適です。
このプログラムをLinux上で長期運用するために：
• systemd（サービス設定ファイル）の書き方や、ログが溢れないようにする設定手順などの解説は必要ですか？
Linux（systemd）上でMQTTサブスクライバを常時動作させ、プログラムが万が一クラッシュした際にも自動再起動させるための.service設定ファイルの作成と登録手順です。
1. systemd 設定ファイルの作成
/etc/systemd/system/ ディレクトリに、新しい設定ファイル（例: mqtt-subscriber.service）を作成します。
bash
sudo nano /etc/systemd/system/mqtt-subscriber.service
コードは注意してご使用ください。
ファイルの内容を以下のように記述します。
ini
[Unit]
Description=MQTT Subscriber with PostgreSQL and SQLite Sync
# ネットワークとMQTTブローカー（ローカルで動いている場合）が起動した後に実行
After=network.target
[Service]
# プログラムを実行する Linux ユーザーとグループ（環境に合わせて変更してください）
User=mqttuser
Group=mqttuser
# ワーキングディレクトリ（SQLiteファイルやPythonスクリプトがある場所）
WorkingDirectory=/opt/mqtt_subscriber
# 実行コマンド（仮想環境 venv を使っている場合はその python パスを指定）
ExecStart=/opt/mqtt_subscriber/.venv/bin/python main.py
# 常時動作（デーモン）のための再起動設定
# プログラムが異常終了、または正常終了しても常に再起動する
Restart=always
# 再起動を試みるまでの待ち時間（秒）
RestartSec=5
# 標準出力と標準エラーの出力先（journaldに送る）
StandardOutput=journal
StandardError=journal
# ログ出力がバッファリングされて遅れるのを防ぐ環境変数
Environment=PYTHONUNBUFFERED=1
[Install]
# OS起動時に自動起動させるための設定
WantedBy=multi-user.target
コードは注意してご使用ください。
2. サービスの有効化と起動
ファイルを保存したら、systemd に設定を反映させてサービスを起動します。
bash
# 1. systemd の設定ファイルを再読み込み
sudo systemctl daemon-reload
# 2. OS起動時に自動でサービスが開始されるように有効化
sudo systemctl enable mqtt-subscriber.service
# 3. サービスを今すぐ起動
sudo systemctl start mqtt-subscriber.service
コードは注意してご使用ください。
3. よく使う管理コマンド
常時稼働の運用中に、状態確認やログ確認でよく使うコマンドです。
• ステータス（動作状態）の確認bash
sudo systemctl status mqtt-subscriber.service
コードは注意してご使用ください。
• リアルタイムのログ確認（Pythonの print 出力）bash
# -u でサービス指定、-f でリアルタイム監視（Ctrl+Cで終了）
sudo journalctl -u mqtt-subscriber.service -f
コードは注意してご使用ください。
• サービスの停止・再起動bash
sudo systemctl stop mqtt-subscriber.service # 停止
sudo systemctl restart mqtt-subscriber.service # 再起動
コードは注意してご使用ください。
Linux運用におけるTips：ログの溢れ対策
Pythonプログラム内の print 出力はすべて Linux の journald（システムログ）に集約されます。
もしMQTTのデータ量が非常に多く、ログファイルがディスクを圧迫しそうな場合は、Pythonコード側で print の代わりに logging モジュール を使い、「エラーや同期発生時のみログを出す（INFOやERRORレベルのみ）」、または「ログを数日分でローテーションする」設計にしておくと、より安全な長期運用が可能です。
これで常時稼働の環境が整います。
• 今回のプログラムを動かすにあたり、Pythonの仮想環境（venv）やログ出力の制御（loggingモジュール化）について、追加で準備や設定方法を確認したい部分はありますか？