Sửa lỗi SQLAlchemy PendingRollbackError: Chấm dứt chu kỳ 'Poisoned Session'

Tại sao cơ sở dữ liệu của bạn lại "ngó lơ" bạn

Hiếm có điều gì gây ức chế hơn một database driver từ chối thực thi truy vấn vì một lỗi xảy ra từ ba bước trước đó. Bạn chạy một truy vấn, nó thất bại, và đột nhiên mọi lời gọi hàm sau đó đều vấp phải một bức tường gạch. Đây chính là PendingRollbackError.

sqlalchemy.exc.PendingRollbackError: Can't reconnect until invalid transaction is rolled back

Nguyên nhân gốc rễ: Session của bạn đã bị "nhiễm độc"

Về bản chất, lỗi này giống như một van ngắt an toàn. Nó xảy ra khi một thao tác trước đó—có thể là vi phạm UniqueConstraint đối với email người dùng hoặc truy vấn bị timeout sau 5 giây—bị thất bại, nhưng session vẫn đang mở.

SQLAlchemy sẽ đánh dấu transaction hiện tại là "inactive" (không hoạt động) hoặc "failed" (thất bại) ngay khi có exception xảy ra. Vì kết nối cơ sở dữ liệu vẫn bị ràng buộc với trạng thái lỗi đó, SQLAlchemy sẽ chặn các lệnh mới để ngăn bạn vô tình commit dữ liệu không hoàn chỉnh hoặc bị hỏng. Nó sẽ không cho phép bạn tiếp tục cho đến khi bạn xác nhận lỗi đó.

Bạn có khả năng sẽ gặp phải tình trạng này trong ba trường hợp cụ thể:

• Rollback bị lãng quên: Bạn đã catch một exception trong khối try/except nhưng không yêu cầu session thực hiện reset.
• Nhiễm chéo giữa các Thread: Bạn đang sử dụng một global session nơi sự cố của một thread làm hỏng kết nối của tất cả các thread khác.
• Các tác vụ "thây ma": Các background worker như Celery đang tái sử dụng một session bị hỏng cho các job khác nhau.

Cách xử lý 1: Lưới an toàn thủ công

Cách khắc phục đáng tin cậy nhất là đảm bảo mọi lệnh session.commit() đều nằm trong khối try/except. Nếu commit thất bại, lệnh rollback() sẽ xóa trạng thái "nhiễm độc" ngay lập tức.

from sqlalchemy.exc import SQLAlchemyError

def create_user_profile(session, profile_data):
    try:
        session.add(profile_data)
        session.commit()
    except SQLAlchemyError as e:
        # Điều này xóa trạng thái lỗi để yêu cầu tiếp theo có thể hoạt động
        session.rollback()
        print(f"Database error: {e}")
        raise
    finally:
        session.close()

Cách xử lý 2: Sử dụng Context Manager hiện đại (Phong cách SQLAlchemy 2.0)

Nếu bạn đang sử dụng SQLAlchemy 1.4 hoặc 2.0, hãy ngừng quản lý commit một cách thủ công. Hãy sử dụng context manager begin(). Nó gọn gàng hơn, an toàn hơn và tự động xử lý rollback cho bạn nếu có bất kỳ sự cố nào xảy ra.

from sqlalchemy.orm import Session

# Session sẽ tự động rollback nếu có lỗi xảy ra bên trong khối 'with'
with Session(engine) as session:
    with session.begin():
        # Giả sử lệnh này gây ra lỗi DuplicateKey error
        session.add(User(email="duplicate@example.com")) 

# Khi thoát khỏi khối lệnh, session sẽ sạch sẽ và sẵn sàng cho tác vụ tiếp theo.

Cách xử lý 3: Tăng cường bảo mật cho FastAPI Dependency

Trong FastAPI, lỗi này thường xuất hiện vì một request thất bại để lại một kết nối "bẩn" trong connection pool. Người dùng tiếp theo nhận được kết nối đó sẽ kế thừa luôn lỗi này. Hãy sử dụng pattern yield để đảm bảo mỗi request đều bắt đầu với một trạng thái mới hoàn toàn.

from fastapi import Depends
from database import SessionLocal

def get_db_session():
    db = SessionLocal()
    try:
        yield db
    except Exception:
        # Nếu route bị crash, chúng ta thực hiện rollback tại đây
        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"}

Cách xử lý 4: An toàn đa luồng với Scoped Sessions

Bạn đang xây dựng một ứng dụng đa luồng? Hãy sử dụng scoped_session. Điều này cung cấp cho mỗi thread một session riêng biệt. Nếu Thread A bị crash, Thread B vẫn tiếp tục chạy mà không bao giờ gặp phải PendingRollbackError.

from sqlalchemy.orm import scoped_session, sessionmaker

session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

# Truy cập session theo từng thread
local_session = Session()
try:
    # Thực hiện các thao tác DB
    local_session.commit()
except:
    local_session.rollback()
finally:
    # Luôn xóa session để trả kết nối về pool
    Session.remove() 

Cách xác minh việc khắc phục

Đừng đoán xem nó đã được sửa hay chưa. Hãy chứng minh điều đó. Chạy một script nhỏ cố tình làm hỏng một transaction, sau đó thử chạy một transaction hợp lệ ngay sau đó.

• Thử chèn một bản ghi có Primary Key bị trùng (ví dụ: ID 101).
• Bắt lỗi IntegrityError.
• Chạy lệnh session.execute("SELECT 1").

Nếu SELECT 1 trả về kết quả, logic rollback của bạn đã ổn định. Nếu nó ném ra lỗi PendingRollbackError, session của bạn vẫn đang bị "nhiễm độc".

# Script kiểm tra nhanh
try:
    session.add(User(id=1)) # Kích hoạt một lỗi crash
    session.commit()
except Exception:
    session.rollback() # Bước khắc phục quan trọng

# Nếu lệnh này hoạt động, bạn đã giải quyết được vấn đề
check = session.execute("SELECT 1").scalar()
print(f"Session Status: {'Healthy' if check == 1 else 'Broken'}")