I'm stuck with asyncio.get_event_loop.

[tommy1511-hub]

Body

I'm stuck with asyncio; my program keeps throwing errors. If anyone knows how to fix it, please help me. Thank you!
import signal
import logging
import schedule
import threading
import time
import asyncio
import os
from config import TELEGRAM_BOT_TOKEN
from telegram import Update
from telegram.ext import Application
from fanpage import monitor_multiple_sheets
from accounts_ID import account_id
from facebook_ads import process_ad_account
from ktra_fb_ads_30p import check_fb_ads_30p

logging.basicConfig(
format="%(asctime)s - %(levelname)s - %(message)s",
level=logging.WARNING # Sử dụng level INFO để dễ theo dõi
)
logger = logging.getLogger(name)

RENDER_APP_URL = "https://fanpage.onrender.com/"
PORT = int(os.environ.get("PORT", 8447))
ENV = "dev" # Chuyển đổi giữa "prod" và "dev" nếu cần

Tạo vòng lặp sự kiện toàn cục

loop = asyncio.get_event_loop()

def handle_exit(_):
logging.info("⏹ Nhận tín hiệu dừng, thoát chương trình...")
# Đóng vòng lặp sự kiện khi dừng chương trình
if not loop.is_closed():
loop.stop()
os._exit(0)

def schedule_tasks():
schedule.every().hour.at(":10").do(ktr_fanpage)
schedule.every().hour.at(":20").do(ktr_fanpage)
schedule.every().hour.at(":30").do(check_campaign_status)
schedule.every().hour.at(":40").do(run_account_id)
schedule.every().hour.at(":50").do(chi_tieu_QC)

# schedule.every().day.at("11:04").do(run_account_id)     
# schedule.every().day.at("11:06").do(chi_tieu_QC)            

def run_account_id():
try:
logging.info("🔄 Đang quét danh sách tài khoản quảng cáo...")
loop.create_task(account_id())
logging.info("✅ Hoàn thành get_ad_accounts()!")
except Exception as e:
logging.error(f"❌ Lỗi khi chạy get_ad_accounts(): {e}")

Hàm ktra fanpage

def ktr_fanpage():
try:
logging.info("🔄 Đang chạy chương trình ktra fanpage...")
loop.create_task(monitor_multiple_sheets())
logging.info("✅ Đã hoàn thành ktra fanpage!")
except Exception as e:
logging.error(f"❌ Lỗi khi chạy chương trình kiểm tra fanpage: {e}")

Hàm ktra trạng thái chiến dịch

def check_campaign_status():
try:
logging.info("🔄 Đang chạy chương trình ktra chiến dịch...")
loop.create_task(check_fb_ads_30p())
logging.info("✅ Đã hoàn thành ktra fanpage!")
except Exception as e:
logging.error(f"❌ Lỗi khi chạy chương trình kiểm tra fanpage: {e}")

def chi_tieu_QC():
try:
logging.info("🔄 Đang chạy chương trình quét chi tiêu chiến dịch...")
loop.create_task(process_ad_account())
logging.info("✅ Đã hoàn thành quét chi tiêu chiến dịch!")
except Exception as e:
logging.error(f"❌ Lỗi khi quét chi tiêu chiến dịch: {e}")

def run_scheduler():
while True:
schedule.run_pending()
time.sleep(30) # Kiểm tra lịch trình sau mỗi 30 giây

def main():
logging.info("🚀 BOT đang khởi chạy trên Render...")

application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()

signal.signal(signal.SIGTERM, handle_exit)
signal.signal(signal.SIGINT, handle_exit)

schedule_tasks()
threading.Thread(target=run_scheduler, daemon=True).start()

if ENV == "dev":
    application.run_polling(allowed_updates=Update.ALL_TYPES)
else:
    application.run_webhook(
        listen="0.0.0.0",
        port=PORT,
        url_path=TELEGRAM_BOT_TOKEN,
        webhook_url=RENDER_APP_URL + TELEGRAM_BOT_TOKEN,
    )

if name == "main":
main()

Guidelines

• I have read and understood this category's guidelines before making this post.

[Kyoka-Reihaido]

think ur having issues with mixing asyncio's event loop with threading in your scheduler. the main problem is that you're creating a global event loop and then trying to use loop.create_task() from different threads, which isn't thread-safe.

Remove global loop = asyncio.get_event_loop()
Add a run_async_function helper that creates a separate event loop for each task
fix your scheduled functions to use this helper with threading
change the main function to use asyncio.run() which is way better

like dis

def run_async_function(coroutine_func):
    async def wrapper():
        try:
            await coroutine_func()
        except Exception as e:
            logging.error(f"❌ Error running async function: {e}")
    
    # new event loop for this thread
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    try:
        loop.run_until_complete(wrapper())
    finally:
        loop.close()

then just call it like this in your scheduled tasks:

def ktr_fanpage():
    try:
        logging.info("🔄 Đang chạy chương trình ktra fanpage...")
        # new thread for the async function
        threading.Thread(target=run_async_function, args=(monitor_multiple_sheets,), daemon=True).start()
        logging.info("✅ Đã hoàn thành ktra fanpage!")
    except Exception as e:
        logging.error(f"❌ Lỗi khi chạy chương trình kiểm tra fanpage: {e}")

basically each thread gets its own event loop and your problem should be fixed. make sure your main also uses asyncio.run(main()) instead of that global loop stuff 🥴

[tommy1511-hub]

I'm also new to programming 😁, thank you very much.

[Kyoka-Reihaido]

Happy to help! ^-^

If that fixed it, please mark the discussion as answered. If ya still have problems, just let me know, and we'll work on a fix together! (๑˃̵ᴗ˂̵)و

[tommy1511-hub]

I have tried adjusting and testing these functions, but my program only runs for a while before encountering an error.
def run_async_function(coroutine_func):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(coroutine_func())
finally:
loop.close()

"Unknown error in HTTP implementation: RuntimeError('Event loop is closed')"