如何通过任务调度程序(aioschedule)运行异步函数？
如何通过任务调度程序(aioschedule)运行异步函数？
提问于 2022-08-24 14:03:30
我正在写一个个人电报机器人，它的一个功能是显示我的帐户market.csgo.com的余额。我的代码：
import asyncio
import aiogram
import aiohttp
...
async def get_balance(session, profiles_dict, message):
async with session.get(f'https://market.csgo.com/api/v2/get-money?key={profiles_dict[1][1]}') as resp:
html = await resp.json()
each_wallet = int(html['money'])
await bot.send_message(message.from_user.id,
f' <a href="{profiles_dict[1][0]}">{profiles_dict[0]}</a> : <i>{each_wallet}</i>',
disable_web_page_preview=True, parse_mode=types.ParseMode.HTML)
...
@dp.message_handler(content_types=['text'])
async def main(message):
profiles = users()
async with aiohttp.ClientSession(trust_env=True) as session:
tasks = []
if message.text == 'Balance ':
await bot.send_message(message.from_user.id, 'Information request. Wait..')
for i in profiles.items():
task = asyncio.ensure_future(get_balance(session, i, message, stats))
tasks.append(task)
await asyncio.gather(*tasks)
if message.text == 'On Sale ':
...
if message.text == 'Timeout Items ⌛':
...
executor.start_polling(dp, skip_updates=False)get_balance()以异步模式工作，向API发送aiohttp请求并输出信息await bot.send_message()。结果：
现在该函数的启动是通过键盘按钮实现的，但是如何使该函数每小时运行一次呢？我知道异步任务调度程序[医]干酪的存在，并且看过这个例子。但是他们运行一个没有参数的函数，但是我有多达3个函数是async def get_balance(session, profiles_dict, message)。我试过这样做：
import asyncio
import aioschedule
async def scheduler(session, profiles_dict, message):
aioschedule.every().hour.do(get_balance(session, profiles_dict, message))
while True:
await aioschedule.run_pending()
await asyncio.sleep(1)
async def on_startup(session, profiles_dict, message):
asyncio.create_task(scheduler(session, profiles_dict, message))
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=False, on_startup=on_startup(session, profiles_dict, message))显然不是这样的。
我的问题是：
如何运行带有参数的异步函数，通过任务调度aiohttp aioschedule 发送请求，并通过电报 aiogram**?**显示结果
Stack Overflow用户
回答已采纳
发布于 2022-08-25 17:15:59
解决方案：
import aiogram
import asyncio
import aiohttp
import aioschedule
...
async def get_balance(session, profiles_dict):
async with session.get(f'https://market.csgo.com/api/v2/get-money?key={profiles_dict[1][1]}') as resp:
html = await resp.json()
each_wallet = int(html['money'])
await bot.send_message(MY_TELEGRAM_ID,
f' <a href="{profiles_dict[1][0]}">{profiles_dict[0]}</a> : <i>{each_wallet}</i>',
disable_web_page_preview=True, parse_mode=types.ParseMode.HTML)
...
@dp.message_handler(content_types=['text'])
async def main(message):
profiles = users()
async with aiohttp.ClientSession(trust_env=True) as session:
tasks = []
if message.text == 'Balance ':
await bot.send_message(message.from_user.id, 'Information request. Wait..')
for i in profiles.items():
task = asyncio.ensure_future(get_balance(session, i))
tasks.append(task)
await asyncio.gather(*tasks)
if message.text == 'On Sale ':
...
if message.text == 'Timeout Items ⌛':
...
# Client session get_balance function
async def session_get_balance():
profiles = users()
async with aiohttp.ClientSession(trust_env=True) as session:
tasks = []
for i in profiles.items():
task = asyncio.ensure_future(get_balance(session, i))
tasks.append(task)
await asyncio.gather(*tasks)
# Schedule functions by time
async def scheduler():
aioschedule.every().hour.do(session_get_balance)
while True:
await aioschedule.run_pending()
await asyncio.sleep(1)
# Function at start
async def on_startup(_):
asyncio.create_task(scheduler())
# Launch telegram bot
if __name__ == "__main__":
executor.start_polling(dp, skip_updates=True, on_startup=on_startup)因为这是我的个人机器人，所以我指定了我的message.from_user.id，而不是MY_TELEGRAM_ID。
await bot.send_message(MY_TELEGRAM_ID,
f' <a href="{profiles_dict[1][0]}">{profiles_dict[0]}</a> : <i>{each_wallet}</i>',
disable_web_page_preview=True, parse_mode=types.ParseMode.HTML)页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/73474566
复制相关文章
相似问题
腾讯云开发者
