python-telegram-bot 20.4异步模式下CustomContext用户认证问题

问题根源 #

python-telegram-bot 20.x中，CustomContext.from_update是同步类方法，框架会同步调用它生成Context实例。如果强行将其改为异步方法，返回的是协程对象而非Context实例，必然触发AttributeError: 'coroutine' object has no attribute 'refresh_data'。

方案1：保留CustomContext，同步方法内执行异步查询 #

不用修改from_update的同步属性，在方法内部通过事件循环执行异步数据库查询。适配Django环境的实现代码如下：

from telegram import Update
from telegram.ext import ContextTypes
import asyncio
from asgiref.sync import sync_to_async
from django.contrib.auth.models import User  # 替换为你的用户模型

# 封装异步用户查询/创建逻辑
async def get_or_create_user(telegram_user_id, telegram_username):
    get_user = sync_to_async(lambda uid: User.objects.get(telegram_id=uid))
    create_user = sync_to_async(lambda uid, un: User.objects.create(telegram_id=uid, username=un))
    
    try:
        return await get_user(telegram_user_id)
    except User.DoesNotExist:
        return await create_user(telegram_user_id, telegram_username)

class CustomContext(ContextTypes.DEFAULT_TYPE):
    def __init__(self, update: Update, application):
        super().__init__(update, application)
        self.db_user = None  # 存储数据库用户对象

    @classmethod
    def from_update(cls, update: Update, application):
        context = cls(update, application)
        if update.effective_user:
            # 同步方法内执行异步任务
            try:
                loop = asyncio.get_running_loop()
                context.db_user = loop.run_until_complete(
                    get_or_create_user(update.effective_user.id, update.effective_user.username)
                )
            except Exception as e:
                # 处理查询异常，比如日志记录
                pass
        return context

如果机器人运行在非ASGI环境，可将loop.run_until_complete替换为asyncio.run。

方案2：改用异步中间件（推荐，贴合框架异步设计） #

利用python-telegram-bot 20.x的异步中间件能力，在请求处理前异步查询用户数据并存入context.user_data，无需自定义Context：

from telegram import Update
from telegram.ext import BaseMiddleware
from asgiref.sync import sync_to_async
from django.contrib.auth.models import User

class UserAuthMiddleware(BaseMiddleware):
    async def pre_process(self, update: Update, context):
        if not update.effective_user:
            return update, context
        
        # 异步查询或创建用户
        try:
            db_user = await sync_to_async(User.objects.get)(telegram_id=update.effective_user.id)
        except User.DoesNotExist:
            db_user = await sync_to_async(User.objects.create)(
                telegram_id=update.effective_user.id,
                username=update.effective_user.username
            )
        # 将用户对象存入user_data，后续handler直接调用
        context.user_data["db_user"] = db_user
        return update, context

# 注册中间件到机器人应用
application = Application.builder().token("YOUR_BOT_TOKEN").build()
application.add_middleware(UserAuthMiddleware())

# 示例handler中使用
async def start_handler(update: Update, context):
    db_user = context.user_data["db_user"]
    await update.message.reply_text(f"欢迎回来，{db_user.username}")

方案3：使用Django 4.2+异步ORM（性能最优） #

Django 4.2原生支持异步ORM操作，可直接替换sync_to_async，代码更简洁高效：

from telegram import Update
from telegram.ext import BaseMiddleware
from django.contrib.auth.models import User

class UserAuthMiddleware(BaseMiddleware):
    async def pre_process(self, update: Update, context):
        if update.effective_user:
            try:
                db_user = await User.objects.aget(telegram_id=update.effective_user.id)
            except User.DoesNotExist:
                db_user = await User.objects.acreate(
                    telegram_id=update.effective_user.id,
                    username=update.effective_user.username
                )
            context.user_data["db_user"] = db_user
        return update, context

内容的提问来源于stack exchange，提问作者Cristian Favaro Carriço