如何在Telegram Bot中实现Django异步数据库调用

问题背景 #

有一个Django应用，Telegram机器人作为自定义命令运行（python manage.py bot）。当用户发送/animals指令时，需要从Animal模型中查询数据并返回，但异步环境下ORM调用一直报错：

• 直接使用Animal.objects.all()时，报错：SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
• 尝试封装成异步函数后，报错：TypeError: 'coroutine' object is not iterable

文件结构：

|Accounts---------------------
|------| models.py------------
|Main-------------------------
|------| Management-----------
|---------------| Commands----
|-----------------------bot.py

Animal模型代码：

class Animal(models.Model):
    id = models.AutoField(primary_key=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    name = models.CharField(max_length=255)

原bot.py关键代码片段：

# 移除了无关导入
from asgiref.sync import sync_to_async
from accounts.models import Animal

class Command(BaseCommand):
    help = "Starts the telegram bot."
    TOKEN = "abc123"

    def handle(self, *args, **options):

        async def get_animals():
            await Animal.objects.all()

        async def animals_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:

            async_db_results = get_animals()

            message = ""
            counter = 0
            for animal in async_db_results:
                message += animal.name + "\n"
                counter += 1

            await update.message.reply_text(message)

        application = Application.builder().token(TOKEN).build()
        application.add_handler(CommandHandler("animals", animals_command))
        application.run_polling(allowed_updates=Update.ALL_TYPES)

错误原因 #

1. Django ORM为同步实现，不能直接在异步上下文调用，必须通过sync_to_async包装同步操作
2. 调用异步函数get_animals()时未添加await关键字，导致拿到的是协程对象而非实际查询结果，无法进行迭代

修复方案 #

步骤1：用sync_to_async包装ORM查询 #

将同步的ORM查询转换为异步可调用函数，同时将QuerySet转换为列表（避免QuerySet惰性求值在异步环境下的问题）

步骤2：正确等待异步函数返回结果 #

调用get_animals()时必须添加await，获取实际查询结果列表后再进行迭代

修复后的bot.py代码 #

# 移除了无关导入
from asgiref.sync import sync_to_async
from accounts.models import Animal

class Command(BaseCommand):
    help = "Starts the telegram bot."
    TOKEN = "abc123"

    def handle(self, *args, **options):
        # 用sync_to_async包装同步ORM查询，转成列表避免惰性求值问题
        @sync_to_async
        def get_animals():
            return list(Animal.objects.all())

        async def animals_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
            # 必须await获取异步函数的返回结果
            animals = await get_animals()
            
            message = ""
            for animal in animals:
                message += animal.name + "\n"

            await update.message.reply_text(message if message else "暂无动物记录")

        application = Application.builder().token(TOKEN).build()
        application.add_handler(CommandHandler("animals", animals_command))
        application.run_polling(allowed_updates=Update.ALL_TYPES)

额外说明 #

• 如果需要过滤当前Telegram用户对应的Animal记录，可以在get_animals中接收用户ID参数，比如从update.effective_user.id获取后传入，修改查询为Animal.objects.filter(user__telegram_id=user_id)（假设User模型有telegram_id字段）
• 原型使用runserver和长轮询无问题，正式环境建议改用生产服务器和Webhook

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