Flask+SQLAlchemy报错：Object已关联到其他Session问题求助

开发一个需登录后报名班次的Flask应用，使用Flask-Login实现登录功能，User和Entry为核心数据模型。尝试将从数据库加载的User对象关联到新创建的Entry时，调用database.py中的insert_entry函数触发以下错误：

sqlalchemy.exc.InvalidRequestError: Object '<User at 0x7ff1ecc5e1d0>' is already attached to session '1' (this is '2')

相关代码 #

main.py #

import locale
import logging
from datetime import date, datetime

import flask
import flask_login
from flask import Flask, render_template, request, flash, redirect, url_for
from flask_login import login_required, login_user, logout_user
from flask_sqlalchemy import SQLAlchemy

from calendartable import createtable, ServiceTable
from database import insert_entry
from database import load_user
from orm import User

logging.basicConfig(level=logging.DEBUG)
locale.setlocale(locale.LC_ALL, 'de_DE.utf8')

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://...'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

app.config.update(LOGIN_DISABLED=True)
app.secret_key = 'very secret key'  # Change this!
database = SQLAlchemy(app)
session = database.session


# prerequisites
d: date = date(day=1, month=11, year=2022)
user = load_user('')


login_manager = flask_login.LoginManager()
login_manager.init_app(app)
login_manager.login_view = "login"


@login_manager.user_loader
def user_loader(user_id):
    u: User = database.load_user(id)
    return u

orm.py #

from flask_login import UserMixin
from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

Base = declarative_base()


class Entry(Base):
    __tablename__ = 'entry'
    id = Column(Integer, primary_key=True)
    day = Column(DateTime(timezone=True), server_default=func.now())
    isdayshift = Column(Boolean, nullable=False, default=True)
    comment = Column(String(100), nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    user_id = Column(Integer, ForeignKey('user.id'))
    user = relationship("User", back_populates="entry", cascade="save-update")

    def __init__(self, day, user, isdayshift, comment):
        self.day = day
        self.user = user
        self.isdayshift = isdayshift
        self.comment = comment


class User(Base, UserMixin):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    lastname = Column(String(100), nullable=False)
    firstname = Column(String(100), nullable=False)
    email = Column(String(100), nullable=False, unique=True)
    password = Column(String(100), nullable=False)
    is_admin = Column(Boolean, nullable=False, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    entry = relationship("Entry", uselist=False, back_populates="user", cascade="merge")

    def __init__(self, lastname, firstname, email, password, is_admin):
        self.lastname = lastname
        self.firstname = firstname
        self.password = password
        self.email = email
        self.is_admin = is_admin

    def isadmin(self):
        return self.admin

    def is_active(self):
        return True

    def get_id(self):
        return self.id

    def is_authenticated(self):
        return True

    def is_anonymous(self):
        return False

database.py #

import logging

from sqlalchemy.ext.declarative import declarative_base

import main
from orm import User, Entry

logging.basicConfig(level=logging.DEBUG)

Base = declarative_base()


def load_user_by_id(user_id):
    u: User = main.session.query(User).filter_by(id == user_id).first()
    return u


def load_user(login):
    logging.debug("loading user from session: %s", main.session)
    u: User = main.session.query(User).filter_by(lastname='xxx').first()
    return u


def load_entry(date, isdayshift):
    # e: Entry = session.query(Entry).filter_by(day=date, isdayshift=isdayshift).first()
    return


def insert_entry(day, user, isdayshift, comment):
    logging.debug("storing entry from session: %s", main.session)
    eq1 = Entry(day, user, isdayshift, comment)
    main.session.add(eq1)
    main.session.commit()

日志信息 #

DEBUG:root:loading user from session: <sqlalchemy.orm.scoping.scoped_session object at 0x7ff1ecd2cbb0>

DEBUG:root:storing entry from session: <sqlalchemy.orm.scoping.scoped_session object at 0x7ff1ecd2cbb0>

问题原因与修复方案 #

核心原因 #

日志显示的是同一个scoped_session外壳对象，但内部实际的数据库会话实例已经切换（比如请求上下文结束后旧会话被销毁，新请求创建了新会话），而之前加载的User对象仍绑定在旧会话上，导致关联新Entry时出现跨会话冲突。同时代码存在几处语法和逻辑错误，加剧了问题。

修复步骤 #

1. 解决跨会话关联问题
   在insert_entry中先将传入的User对象合并到当前会话，再关联Entry：

   def insert_entry(day, user, isdayshift, comment):
       logging.debug("storing entry from session: %s", main.session)
       # 将user合并到当前会话，避免跨会话冲突
       user = main.session.merge(user)
       eq1 = Entry(day, user, isdayshift, comment)
       main.session.add(eq1)
       main.session.commit()
   

2. 修正用户加载函数的语法错误

   • 修改load_user_by_id的过滤条件（==改为=）：

     def load_user_by_id(user_id):
         u: User = main.session.query(User).filter_by(id=user_id).first()
         return u
     

   • 修复user_loader的错误调用：

     @login_manager.user_loader
     def user_loader(user_id):
         return load_user_by_id(user_id)
     

3. 删除全局提前加载的User对象
   移除main.py中全局作用域的user = load_user('')，在需要加载用户的请求处理函数内再调用加载逻辑，确保用户对象绑定在当前请求的会话上。

4. 修复User类的方法错误
   修正isadmin方法的属性引用：

   def isadmin(self):
       return self.is_admin
   

内容的提问来源于stack exchange，提问作者Norbert