データベースが「応答不能」になる理由
3ステップ前のミスが原因でクエリの実行を拒否するデータベースドライバーほど、イライラするものはありません。クエリを実行して失敗すると、突然それ以降のすべての呼び出しが遮断されてしまいます。これが PendingRollbackError です。
sqlalchemy.exc.PendingRollbackError: Can't reconnect until invalid transaction is rolled back
根本原因：セッションの「汚染」
本質的に、このエラーは安全のための遮断弁のようなものです。これは、ユーザーのメールアドレスにおける UniqueConstraint（一意制約）違反や5秒のクエリタイムアウトなど、以前の操作が失敗したにもかかわらず、セッションが開いたままになっている場合に発生します。
SQLAlchemyは、例外が発生した瞬間に現在のトランザクションを「非アクティブ」または「失敗」としてマークします。データベース接続がその失敗した状態に紐付いたままであるため、SQLAlchemyは不完全なデータや破損したデータを誤ってコミットしないよう、新しいコマンドをブロックします。失敗を認識（ロールバック）するまで、処理を先に進めることはできません。
このエラーは、主に以下の3つの状況で発生します：
- ロールバックの忘れ：
try/exceptブロックで例外をキャッチしたものの、セッションのリセットを指示しなかった場合。 - スレッド間の干渉： グローバルセッションを使用しており、あるスレッドのクラッシュが他のすべてのスレッドの接続を壊してしまう場合。
- ゾンビタスク： Celeryなどのバックグラウンドワーカーが、異なるジョブ間で壊れたセッションを再利用している場合。
解決策1：手動によるセーフティネット
最も確実な解決策は、すべての session.commit() を try/except ブロック内で行うことです。コミットが失敗した場合、rollback() を実行することで「汚染された」状態を即座に解消できます。
from sqlalchemy.exc import SQLAlchemyError
def create_user_profile(session, profile_data):
try:
session.add(profile_data)
session.commit()
except SQLAlchemyError as e:
# 失敗した状態をクリアし、次のリクエストが動作するようにします
session.rollback()
print(f"Database error: {e}")
raise
finally:
session.close()
解決策2：モダンなコンテキストマネージャー（SQLAlchemy 2.0 スタイル）
SQLAlchemy 1.4 または 2.0 を使用している場合は、手動でコミットを管理するのはやめましょう。begin() コンテキストマネージャーを使用してください。この方がコードがすっきりし、より安全で、問題が発生した際のロールバックも自動的に処理されます。
from sqlalchemy.orm import Session
# 'with' ブロック内でエラーが発生すると、セッションは自動的にロールバックされます
with Session(engine) as session:
with session.begin():
# ここで DuplicateKey エラーが発生すると仮定します
session.add(User(email="duplicate@example.com"))
# ブロックを抜けると、セッションはクリーンになり、次のタスクの準備が整います。
解決策3：FastAPIの依存関係を強固にする
FastAPIでは、リクエストの失敗によってコネクションプール内の接続が「汚れた」まま放置されることが原因で、このエラーがよく発生します。その接続を次に受け取ったユーザーがエラーを引き継いでしまうのです。yield パターンを使用して、すべてのリクエストがクリーンな状態で開始されるようにしましょう。
from fastapi import Depends
from database import SessionLocal
def get_db_session():
db = SessionLocal()
try:
yield db
except Exception:
# ルートでクラッシュが発生した場合、ここでロールバックします
db.rollback()
raise
finally:
db.close()
@app.post("/register")
def register_user(user: UserSchema, db: Session = Depends(get_db_session)):
db.add(user)
db.commit()
return {"message": "User created"}
解決策4：Scoped Sessionによるスレッドセーフの確保
マルチスレッドアプリケーションを構築していますか？その場合は scoped_session を使用してください。これにより、各スレッドに個別の隔離されたセッションが割り当てられます。スレッドAがクラッシュしても、スレッドBは PendingRollbackError に遭遇することなく実行を継続できます。
from sqlalchemy.orm import scoped_session, sessionmaker
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)
# スレッドローカルなセッションへのアクセス
local_session = Session()
try:
# DB操作を実行
local_session.commit()
except:
local_session.rollback()
finally:
# セッションを必ず削除し、接続をプールに返却します
Session.remove()
修正を確認する方法
修正されたかどうかを推測するのではなく、証明しましょう。意図的にトランザクションを失敗させた直後に、有効なトランザクションを実行する小さなスクリプトを走らせます。
- 重複する主キー（例：ID 101）を持つレコードを挿入してみる。
IntegrityErrorをキャッチする。session.execute("SELECT 1")を実行する。
If SELECT 1 が結果を返せば、ロールバックのロジックは確実です。もし PendingRollbackError がスローされるなら、セッションはまだ汚染されたままです。
# クイックテスト用スクリプト
try:
session.add(User(id=1)) # クラッシュを発生させる
session.commit()
except Exception:
session.rollback() # 極めて重要な修正
# これが動作すれば解決です
check = session.execute("SELECT 1").scalar()
print(f"Session Status: {'Healthy' if check == 1 else 'Broken'}")