You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.10 14:00:28