demand_server/app/scheduler/manager.py

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

from app.core.config import settings
from app.scheduler.jobs import (
    demand_pool_daily_strategy_alert_job,
    demand_pool_today_incremental_sync_job,
    heartbeat_job,
)


scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")


def setup_jobs() -> None:
    scheduler.add_job(
        heartbeat_job,
        trigger=IntervalTrigger(seconds=settings.scheduler_heartbeat_seconds),
        id="heartbeat_job",
        replace_existing=True,
    )
    if settings.demand_pool_hourly_sync_enabled:
        scheduler.add_job(
            demand_pool_today_incremental_sync_job,
            trigger=CronTrigger(minute=settings.demand_pool_hourly_sync_minute),
            id="demand_pool_hourly_sync_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
    if settings.demand_pool_daily_strategy_alert_enabled:
        scheduler.add_job(
            demand_pool_daily_strategy_alert_job,
            trigger=CronTrigger(
                hour=settings.demand_pool_daily_strategy_alert_hour,
                minute=settings.demand_pool_daily_strategy_alert_minute,
            ),
            id="demand_pool_daily_strategy_alert_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )


def start_scheduler() -> None:
    if scheduler.running:
        return
    setup_jobs()
    scheduler.start()


def shutdown_scheduler() -> None:
    if not scheduler.running:
        return
    scheduler.shutdown(wait=False)