SQLAlchemy双向多对多关联对象模式引发ArgumentError求助

问题背景 #

尝试通过Cart表作为关联对象，在User与Product之间建立带额外字段的双向多对多关系，但遇到查询错误和关系配置问题。

模型定义代码 #

class Cart(BaseModel):
    __tablename__ = 'cart'
    product_id: Mapped[int] = Column(Integer, ForeignKey('product.id'), primary_key=True)
    user_id: Mapped[int] = Column(Integer, ForeignKey('user.id'), primary_key=True)
    selling_price: Mapped[Decimal] = mapped_column(Numeric(precision=6, scale=2), nullable=True)
    quantity: Mapped[int]
    user: Mapped["User"] = relationship("Users", back_populates="product")
    product: Mapped["Product"] = relationship("Product", back_populates="user")


class User(BaseModel, UserMixin):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    firstname: Mapped[str] = mapped_column(String(20), unique=False, nullable=False)
    lastname: Mapped[str] = mapped_column(String(20), unique=False, nullable=False)
    email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
    password: Mapped[str] = mapped_column(String(60), nullable=False)
    products: Mapped[list['Product']] = relationship("Cart", back_populates='user')

    def add_to_cart(self, product_id: int):
        item_to_add = Cart.insert().values(product_id=product_id, user_id=self.id)
        self.session.execute(item_to_add)
        self.session.commit()
        flash('Your item has been added to your cart!', 'success')

    def __repr__(self):
        return f"User('{self.firstname}','{self.lastname}', '{self.email}','{self.id}')"


class Product(BaseModel):
    __tablename__ = "product"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(precision=6, scale=2), nullable=True)
    description: Mapped[str] = mapped_column(Text(), nullable=False)
    users: Mapped[list['User']] = relationship("Cart", back_populates='product')

    def __repr__(self):
        return f"Product('{self.name}', '{self.price}')"

触发错误的查询代码 #

q = select(User).where(User.email == form.email.data)
user = conn.session.query(q).scalars().first()

报错堆栈 #

Traceback (most recent call last):
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 1478, in __call__
    return self.wsgi_app(environ, start_response)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 1458, in wsgi_app
    response = self.handle_exception(e)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 1455, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 869, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 867, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/flask/app.py", line 852, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "/Users/sholden/Projects/Python/ecommerce/unwrap/routes.py", line 54, in login
    user = conn.session.query(q).scalars().first()
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2912, in query
    return self._query_cls(entities, self, **kwargs)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 276, in __init__
    self._set_entities(entities)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 288, in _set_entities
    self._raw_columns = [
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 289, in <listcomp>
    coercions.expect(
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/sql/coercions.py", line 442, in expect
    return impl._implicit_coercions(
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/sql/coercions.py", line 506, in _implicit_coercions
    self._raise_for_expected(element, argname, resolved)
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/sql/coercions.py", line 1137, in _raise_for_expected
    return super()._raise_for_expected(
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/sql/coercions.py", line 710, in _raise_for_expected
    super()._raise_for_expected(
  File "/Users/sholden/.venvs/ecommerce-3.10/lib/python3.10/site-packages/sqlalchemy/sql/coercions.py", line 535, in _raise_for_expected
    raise exc.ArgumentError(msg, code=code) from err
sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns
clause element expected, got <sqlalchemy.sql.selectable.Select object at 0x104c04d90>.
To create a FROM clause from a <class 'sqlalchemy.sql.selectable.Select'> object, use the
.subquery() method.

错误原因与修复方案 #

1. 查询语句的基础错误 #

session.query()是SQLAlchemy 1.x的旧式API，无法直接传入select()构造的查询对象。正确写法是用session.execute()执行查询：

q = select(User).where(User.email == form.email.data)
result = conn.session.execute(q)
user = result.scalars().first()

2. 关联对象的关系配置错误 #

关联对象模式下，User和Product的关系需正确指向Cart表，且back_populates要一一对应：

• Cart类中user关系的关联类名应为User（而非错误的Users），back_populates对应User类的carts属性
• User类需先定义与Cart的直接关系，若要直接访问关联的Product，可添加viewonly=True的辅助关系
• Product类同理，先定义与Cart的直接关系，再添加与User的辅助关系

修正后的Cart类：

class Cart(BaseModel):
    __tablename__ = 'cart'
    product_id: Mapped[int] = Column(Integer, ForeignKey('product.id'), primary_key=True)
    user_id: Mapped[int] = Column(Integer, ForeignKey('user.id'), primary_key=True)
    selling_price: Mapped[Decimal] = mapped_column(Numeric(precision=6, scale=2), nullable=True)
    quantity: Mapped[int]
    user: Mapped["User"] = relationship("User", back_populates="carts")
    product: Mapped["Product"] = relationship("Product", back_populates="carts")

修正后的User类：

class User(BaseModel, UserMixin):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    firstname: Mapped[str] = mapped_column(String(20), unique=False, nullable=False)
    lastname: Mapped[str] = mapped_column(String(20), unique=False, nullable=False)
    email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
    password: Mapped[str] = mapped_column(String(60), nullable=False)
    carts: Mapped[list['Cart']] = relationship(back_populates='user')
    # 可选：直接访问关联的Product
    products: Mapped[list['Product']] = relationship(
        secondary='cart',
        viewonly=True,
        back_populates='users'
    )

    def add_to_cart(self, product_id: int):
        # 用ORM方式创建对象，更符合ORM设计逻辑
        cart_item = Cart(user_id=self.id, product_id=product_id)
        self.session.add(cart_item)
        self.session.commit()
        flash('商品已加入购物车！', 'success')

修正后的Product类：

class Product(BaseModel):
    __tablename__ = "product"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(precision=6, scale=2), nullable=True)
    description: Mapped[str] = mapped_column(Text(), nullable=False)
    carts: Mapped[list['Cart']] = relationship(back_populates='product')
    # 对应User的products属性
    users: Mapped[list['User']] = relationship(
        secondary='cart',
        viewonly=True,
        back_populates='products'
    )

3. add_to_cart方法优化 #

原代码使用Core层的Cart.insert()写法，ORM中更推荐创建Cart对象并添加到session，能更好利用ORM的关系管理和自动持久化特性。

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