""" Background Task Scheduler for POS Service Sets up periodic background jobs for: - Syncing POS transactions to sales service - Other maintenance tasks as needed To enable scheduling, add to main.py startup: ```python from app.scheduler import start_scheduler, shutdown_scheduler @app.on_event("startup") async def startup_event(): start_scheduler() @app.on_event("shutdown") async def shutdown_event(): shutdown_scheduler() ``` """ import structlog from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from datetime import datetime logger = structlog.get_logger() # Global scheduler instance scheduler = None def start_scheduler(): """ Initialize and start the background scheduler Jobs configured: - POS to Sales Sync: Every 5 minutes """ global scheduler if scheduler is not None: logger.warning("Scheduler already running") return try: scheduler = AsyncIOScheduler() # Job 1: Sync POS transactions to sales service from app.jobs.sync_pos_to_sales import run_pos_to_sales_sync scheduler.add_job( run_pos_to_sales_sync, trigger=IntervalTrigger(minutes=5), id='pos_to_sales_sync', name='Sync POS Transactions to Sales', replace_existing=True, max_instances=1, # Prevent concurrent runs coalesce=True, # Combine multiple missed runs into one misfire_grace_time=60 # Allow 60 seconds grace for missed runs ) scheduler.start() logger.info("Background scheduler started", jobs=len(scheduler.get_jobs()), next_run=scheduler.get_jobs()[0].next_run_time if scheduler.get_jobs() else None) except Exception as e: logger.error("Failed to start scheduler", error=str(e), exc_info=True) scheduler = None def shutdown_scheduler(): """Gracefully shutdown the scheduler""" global scheduler if scheduler is None: logger.warning("Scheduler not running") return try: scheduler.shutdown(wait=True) logger.info("Background scheduler stopped") scheduler = None except Exception as e: logger.error("Failed to shutdown scheduler", error=str(e), exc_info=True) def get_scheduler_status(): """ Get current scheduler status Returns: Dict with scheduler info and job statuses """ if scheduler is None: return { "running": False, "jobs": [] } jobs = [] for job in scheduler.get_jobs(): jobs.append({ "id": job.id, "name": job.name, "next_run": job.next_run_time.isoformat() if job.next_run_time else None, "trigger": str(job.trigger) }) return { "running": True, "jobs": jobs, "state": scheduler.state } def trigger_job_now(job_id: str): """ Manually trigger a scheduled job immediately Args: job_id: Job identifier (e.g., 'pos_to_sales_sync') Returns: True if job was triggered, False otherwise """ if scheduler is None: logger.error("Cannot trigger job, scheduler not running") return False try: job = scheduler.get_job(job_id) if job: scheduler.modify_job(job_id, next_run_time=datetime.now()) logger.info("Job triggered manually", job_id=job_id) return True else: logger.warning("Job not found", job_id=job_id) return False except Exception as e: logger.error("Failed to trigger job", job_id=job_id, error=str(e)) return False