基于PostgreSQL+asyncpg的Telegram Bot查询报错：'NoneType'无'send'属性

我开发了一个基于PostgreSQL和asyncpg的Telegram Bot，数据库新增操作正常，但实现订阅查询功能时出现异常。

订阅新增函数 #

async def add_subscription(user_id: BigInteger, start_date: DateTime, end_date: DateTime):
    async with async_session() as session:
        async with session.begin():
            new_subscription = UserSubscription(
                user_id=user_id,
                subscription_start=start_date,
                subscription_end=end_date
            )
            session.add(new_subscription)
            await session.commit()

订阅查询函数 #

async def get_subscription(user_id: BigInteger):
    async with async_session() as session:
        print(f"Running DB query for user_id: {user_id}")
        async with session.begin():
            print("session begins")
            try:
                query = text("SELECT subscription_start, subscription_end FROM users WHERE user_id = :user_id")
                result = await session.scalars(query, {"user_id": user_id})
                subscription = result._fetchone_impl()
                return subscription
            except Exception as e:
                print(f"Error during DB query execution: {e}")
                return None

执行报错日志 #

session begins
2024-09-04 01:08:52,132 INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
2024-09-04 01:08:52,133 INFO sqlalchemy.engine.Engine SELECT subscription_start, subscription_end FROM users WHERE user_id = $1
INFO:sqlalchemy.engine.Engine:SELECT subscription_start, subscription_end FROM users WHERE user_id = $1
2024-09-04 01:08:52,134 INFO sqlalchemy.engine.Engine [cached since 21.7s ago] (1867377574,)
INFO:sqlalchemy.engine.Engine:[cached since 21.7s ago] (1867377574,)
Error during DB query execution: 'NoneType' object has no attribute 'send'
2024-09-04 01:08:52,135 INFO sqlalchemy.engine.Engine COMMIT
INFO:sqlalchemy.engine.Engine:COMMIT
Subscription: None

Bot中实现了/subscription命令，调用查询函数后返回“未找到您的订阅信息”，同时触发上述报错。

测试逻辑 #

async def main():
    
    # 获取订阅
    subscription = await get_subscription(user_id=1867377574)
    print(subscription)

# 运行主异步循环
asyncio.run(main())

问题分析与解决 #

核心问题1：调用SQLAlchemy内部私有方法

查询函数中使用了result._fetchone_impl()，这是SQLAlchemy的私有内部方法，不属于公开API范畴。当查询结果为空时，内部游标对象可能为None，调用该方法会触发'NoneType' object has no attribute 'send'错误。

核心问题2：表名可能不匹配

新增操作使用的是UserSubscription模型，但查询SQL中指定的表是users。如果UserSubscription对应的数据库表不是users（通常SQLAlchemy会默认生成user_subscription或user_subscriptions这类表名），会导致查询不到数据。

优化后的查询函数（方案一：修正原生SQL查询）

async def get_subscription(user_id: BigInteger):
    async with async_session() as session:
        print(f"Running DB query for user_id: {user_id}")
        async with session.begin():
            print("session begins")
            try:
                # 替换为UserSubscription对应的真实表名，比如user_subscriptions
                query = text("SELECT subscription_start, subscription_end FROM user_subscriptions WHERE user_id = :user_id")
                result = await session.scalars(query, {"user_id": user_id})
                # 使用公开API first()获取单行结果，自动处理空结果
                subscription = result.first()
                return subscription
            except Exception as e:
                print(f"Error during DB query execution: {e}")
                return None

优化后的查询函数（方案二：使用ORM查询更可靠）

推荐直接使用SQLAlchemy ORM查询，避免手写SQL的表名、字段名错误：

async def get_subscription(user_id: BigInteger):
    async with async_session() as session:
        print(f"Running DB query for user_id: {user_id}")
        async with session.begin():
            print("session begins")
            try:
                # 直接通过模型查询，自动对应表和字段
                subscription = await session.get(UserSubscription, user_id)
                # 按需返回指定字段
                if subscription:
                    return (subscription.subscription_start, subscription.subscription_end)
                return None
            except Exception as e:
                print(f"Error during DB query execution: {e}")
                return None

额外优化：新增函数的冗余代码

新增函数中的await session.commit()是冗余的，async with session.begin()上下文管理器会自动处理事务提交/回滚，可删除该语句：

async def add_subscription(user_id: BigInteger, start_date: DateTime, end_date: DateTime):
    async with async_session() as session:
        async with session.begin():
            new_subscription = UserSubscription(
                user_id=user_id,
                subscription_start=start_date,
                subscription_end=end_date
            )
            session.add(new_subscription)
            # 移除冗余的commit
            # await session.commit()

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