提问人:afaf12 提问时间:10/30/2023 更新时间:11/1/2023 访问量:39
如何在 Telegram 机器人中异步进行数据库调用?
How do I make database calls asynchronous inside Telegram bot?
问:
我有一个 Django 应用程序,它以命令的形式运行 Telegram 聊天机器人脚本。
我用 启动 Django 应用程序。
我用 . 启动电报客户端。python manage.py runserver
python manage.py bot
我想在用户在电报聊天中键入“/animals”时调用的异步方法中列出 Animal 表中的条目。如果我使用硬编码列表或字典作为数据源,我的代码就可以工作。但是,我无法让 ORM 调用在异步模式下工作。
文件结构:
|Accounts---------------------
|------| models.py------------
|Main-------------------------
|------| Management-----------
|---------------| Commands----
|-----------------------bot.py
动物模型:
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
# removed unrelated imports
from asgiref.sync import sync_to_async
from accounts.models import Animal
class Command(BaseCommand):
help = "Starts the telegram bot."
# assume that the token is correct
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)
此代码的错误消息:
TypeError: 'coroutine' object is not iterable
最初,我用 .ORM调用不是异步的,所以我收到以下错误消息:Animal.objects.all()
async_db_results
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
这是一个原型应用程序,我知道我不应该使用 .我还应该使用 webhook 而不是长轮询,但我认为这些问题与我的异步问题无关。
我要尝试的下一件事是使用 asyncio,但我已经花了很多时间,我想我会问这个问题。runserver
我查看了这些资源(以及许多其他资源):
docs.djangoproject.com: asgiref.sync.sync_to_async
stackoverflow: Django: SynchronousOnlyOperation: 你不能从异步上下文调用它 - 使用线程或sync_to_async
stackoverflow: 同步到异步 Django ORM 查询集外键属性
答:
我想通了。正确使用并做到了。sync_to_async
await
@sync_to_async
def get_animals():
return list(Animal.objects.all())
async def animals_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Send a message when the command /animal is issued."""
async_db_results = await get_animals()
message = "Animal list:"
for h in async_db_results:
message += "{}\n".format(h.name)
await update.message.reply_text(message)
评论