Coverage for app / services / scheduler.py: 0%

1"""Background scheduler for polling campaign balances.""" 

2from apscheduler.schedulers.asyncio import AsyncIOScheduler 

3from apscheduler.triggers.interval import IntervalTrigger 

4from app.core.config import settings 

5from app.db.database import AsyncSessionLocal 

6from app.services.balance_tracker import BalanceTracker 

7 

8 

9scheduler = AsyncIOScheduler() 

10 

11 

12async def poll_balances_job(): 

13 """Background job to poll all active campaign balances.""" 

14 async with AsyncSessionLocal() as db: 

15 tracker = BalanceTracker(db) 

16 await tracker.poll_all_active_campaigns() 

17 

18 

19def start_scheduler(): 

20 """Start the background scheduler.""" 

21 if scheduler.running: 

22 return 

23 

24 interval_seconds = settings.balance_poll_interval_seconds 

25 scheduler.add_job( 

26 poll_balances_job, 

27 trigger=IntervalTrigger(seconds=interval_seconds), 

28 id="poll_balances", 

29 name="Poll campaign balances", 

30 replace_existing=True, 

31 ) 

32 scheduler.start() 

33 

34 

35def shutdown_scheduler(): 

36 """Shutdown the scheduler.""" 

37 if scheduler.running: 

38 scheduler.shutdown()