SQLAlchemy无异常抛出:子类追加至父类失败排查求助
问题描述
尝试将CardBase实例追加到UserBase实例的关联集合中,代码既不抛出异常也无法正常运行。多款IDE(Spyder、PyCharm、VSC)均未检测到语法错误,显式打印异常的操作大多无效,甚至设置的print语句都无法执行。
相关代码
database_append_card.py
async def append_all(message: types.Message, state: FSMContext): async with state.proxy() as data: new_card = CardBase( name=data['card_name'], front=data['front'], back=data['back'], ) await add_child_to_db( child=new_card, column=str(message.from_user.id), parent_class=UserBase, my_async_session=async_session_maker) await bot.send_message(message.from_id, 'The card has been appended! ✅')
database_commands.py
async def add_child_to_db( child, column, parent_class, my_async_session: AsyncSession): """ Adds a child class to parent class """ async with my_async_session.begin() as session: try: parent = await session.execute(select(parent_class).where(parent_class.column==column)) print(f'\n\n\n\n{parent}\n\n\n\n') parent.children.append(child) except SQLAlchemyError as exc: print(exc) raise finally: await session.close()
database_models.py
class UserBase(Base): """ An account for storing and accessing multiple learning cards """ __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) telegram_id = Column(String, unique=True) username = Column(String(100), unique=True) name = Column(String(200)) surname = Column(String(200)) my_cards: Mapped[list['CardBase']] = relationship() class CardBase(Base): """ A learning card with front and back text bound to a specific user """ __tablename__ = 'cards' id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'), nullable=False) name = mapped_column(Text) front = mapped_column(Text) back = mapped_column(Text)
环境信息
asyncpg==0.28.0 SQLAlchemy==2.0.19 Python 3.10.12
问题修复方案
1. 修正查询结果处理逻辑
session.execute()返回的是Result对象,不是直接的实体实例,且原代码中parent_class.column是错误字段(UserBase无此属性),需替换为telegram_id并正确获取实例:
# 替换原查询代码 result = await session.execute(select(parent_class).where(parent_class.telegram_id == column)) parent = result.scalar_one_or_none() # 增加用户不存在的判断 if not parent: raise ValueError("目标用户不存在")
2. 完善关联关系配置
UserBase和CardBase的关联关系需补充双向绑定和级联规则:
# 修改UserBase的关系定义 my_cards: Mapped[list['CardBase']] = relationship(back_populates="user", cascade="all, delete-orphan") # 在CardBase中添加反向关联 user: Mapped['UserBase'] = relationship(back_populates="my_cards")
3. 优化会话管理
使用async with my_async_session.begin()时,上下文管理器会自动处理会话关闭,无需手动调用session.close(),可删除finally块中的关闭操作。
4. 修正关联集合调用
原代码中parent.children是错误的,需替换为UserBase中定义的my_cards属性:
parent.my_cards.append(child)
内容的提问来源于stack exchange,提问作者Ani
相关产品推荐
相关产品推荐

