迁移至PostgreSQL后find_and_lock_token方法报NoneType错误求助

报错'NoneType' object has no attribute 'fetchrow'的核心是调用fetchrow的对象为None，尽管你确认初始连接正常，但结合异步场景与代码逻辑，以下是可能的原因及解决方法：

可能的触发原因 #

• 异步初始化时序问题：若代码在其他异步任务中提前调用find_and_lock_token，或全局变量conn在异步上下文里被意外重置，会导致传入方法的连接对象无效。
• 连接中途断开：循环等待（asyncio.sleep(5)）过程中，数据库连接可能因超时、服务重启等原因断开，使conn变为None。
• 全局变量竞态：全局连接在多异步任务共享时，可能被其他操作意外修改为无效状态。

针对性解决方法 #

1. 改用连接池替代全局连接（推荐） #

全局连接在异步场景下易出现状态不一致问题，使用asyncpg连接池可确保每次获取的都是有效连接，同时优化并发处理：

async def init_db_pool():
    return await asyncpg.create_pool(
        user=await get_username(),
        password=await get_pass(),
        host=await get_host(),
        port=await get_port(),
        database=await get_username()
    )

async def find_and_lock_token(pool):
    while True:
        async with pool.acquire() as conn:
            try:
                # 原子化完成查询+更新，避免并发冲突
                token = await conn.fetchval(
                    '''UPDATE "Tokens" 
                       SET "Status" = $1 
                       WHERE "Token" = (
                           SELECT "Token" FROM "Tokens" WHERE "Status" = $2 LIMIT 1 FOR UPDATE SKIP LOCKED
                       )
                       RETURNING "Token"''',
                    'Busy', 'Free'
                )
                if token is not None:
                    return token
                await asyncio.sleep(5)
            except Exception as e:
                logging.error(f"Error finding and locking token: {e}")

# 使用示例
pool = await init_db_pool()
token = await find_and_lock_token(pool)

2. 检查连接有效性并自动重连 #

若必须使用单连接，在循环内添加连接状态检查，无效则重新初始化：

async def find_and_lock_token():
    global conn
    while True:
        try:
            # 校验连接是否有效，无效则重建
            if conn is None or conn.is_closed():
                await init_db_connection()
            
            token_row = await conn.fetchrow(
                'SELECT "Token" FROM "Tokens" WHERE "Status" = $1 LIMIT 1 FOR UPDATE SKIP LOCKED',
                'Free'
            )

            if token_row is not None:
                token_val = token_row['Token']
                await conn.execute(
                    'UPDATE "Tokens" SET "Status" = $1 WHERE "Token" = $2',
                    'Busy', token_val
                )
                return token_val
            await asyncio.sleep(5)
        except Exception as e:
            logging.error(f"Error finding and locking token: {e}")
            # 出错后重置连接，避免后续循环使用无效对象
            conn = None
            await asyncio.sleep(2)

3. 原子化SQL操作优化 #

原代码中SELECT与UPDATE分离，高并发场景下可能出现多进程同时获取同一Free令牌的问题。改用UPDATE ... RETURNING可原子完成查询与锁定，既解决并发冲突，也简化代码逻辑。

额外验证点 #

• 确认PostgreSQL表名、列名大小写：PostgreSQL对标识符大小写敏感，若建表时用小写，代码中双引号包裹的大写会导致对象找不到（虽非直接报错原因，但需排查）。
• 校验get_username()等方法返回的数据库参数是否正确，确保连接目标数据库。

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