Pranav/cron job (#59)

* initial structure of mark as complete service

* added infrastructure for cron jobs + implemented trial job

* configure timezone appropriately

* linting + formatting

* fix lint issues

* copilot review fixes

* linter fixes

Author: PranavGopinath

backend/python/app/__init__.py

@@ -6,6 +6,9 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from app.dependencies.services import get_scheduler_service
+from app.services.jobs import init_jobs
+
 from .config import settings
 from .models import init_app as init_models
 from .routers import init_app as init_routers
@@ -119,11 +122,15 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     initialize_firebase()
     init_models()
 
+    # Initialize scheduler
+    scheduler_service = get_scheduler_service()
+    scheduler_service.start()
+    init_jobs(scheduler_service)
+
     yield
 
-    # Shutdown
-    # Add any cleanup code here if needed
-    pass
+    # Cleanup: stop the scheduler service during application shutdown
+    scheduler_service.stop()
 
 
 def create_app() -> FastAPI:

backend/python/app/config.py

@@ -57,6 +57,9 @@ class Settings(BaseSettings):
     port: int = Field(default=8080)
     host: str = Field(default="0.0.0.0")
 
+    # Scheduler
+    scheduler_timezone: str = Field(default="America/New_York")
+
     # Preview deploy
     preview_deploy: bool = Field(default=False)
 

backend/python/app/dependencies/services.py

@@ -20,6 +20,7 @@
     MockRoutingAlgorithm,
 )
 from app.services.implementations.route_group_service import RouteGroupService
+from app.services.implementations.scheduler_service import SchedulerService
 from app.services.implementations.simple_entity_service import SimpleEntityService
 from app.services.protocols.routing_algorithm import RoutingAlgorithmProtocol
 
@@ -105,3 +106,10 @@ def get_routing_algorithm() -> RoutingAlgorithmProtocol:
     Swap this to use a different algorithm implementation.
     """
     return MockRoutingAlgorithm()
+
+
+@lru_cache
+def get_scheduler_service() -> SchedulerService:
+    """Get scheduler service instance"""
+    logger = get_logger()
+    return SchedulerService(logger)

backend/python/app/services/implementations/scheduler_service.py

@@ -0,0 +1,128 @@
+import asyncio
+import logging
+from collections.abc import Callable
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from apscheduler.schedulers.background import (  # type: ignore[import-untyped]
+    BackgroundScheduler,
+)
+from apscheduler.triggers.cron import CronTrigger  # type: ignore[import-untyped]
+
+from app.config import settings
+
+
+class SchedulerService:
+    """Centralized service for managing scheduled tasks (cron jobs)"""
+
+    def __init__(self, logger: logging.Logger):
+        self.logger = logger
+        self.scheduler: BackgroundScheduler | None = None
+        self._is_running = False
+        # Get timezone from settings
+        self.timezone = ZoneInfo(settings.scheduler_timezone)
+
+    def start(self) -> None:
+        """Start the scheduler"""
+        if self._is_running:
+            self.logger.warning("Scheduler is already running")
+            return
+
+        # Pass timezone to scheduler (applies to all jobs)
+        self.scheduler = BackgroundScheduler(timezone=self.timezone)
+        self.scheduler.start()
+        self._is_running = True
+        self.logger.info(f"Scheduler started with timezone: {self.timezone}")
+
+    def stop(self) -> None:
+        """Stop the scheduler"""
+        if not self._is_running or self.scheduler is None:
+            return
+
+        self.scheduler.shutdown(wait=True)
+        self._is_running = False
+        self.logger.info("Scheduler stopped")
+
+    def add_cron_job(
+        self,
+        func: Callable,
+        job_id: str,
+        hour: int | str = "*",
+        minute: int | str = "*",
+        day_of_week: int | str = "*",
+        day: int | str = "*",
+        month: int | str = "*",
+    ) -> None:
+        """Add a cron job to the scheduler
+
+        Args:
+            func: The function to execute (can be sync or async)
+            job_id: Unique identifier for the job
+            hour: Hour (0-23) or '*' for every hour (in configured timezone)
+            minute: Minute (0-59) or '*' for every minute
+            day_of_week: Day of week (0-6, 0=Monday) or '*' for every day
+            day: Day of month (1-31) or '*' for every day
+            month: Month (1-12) or '*' for every month
+        """
+        if not self._is_running or self.scheduler is None:
+            raise RuntimeError("Scheduler must be started before adding jobs")
+
+        # Wrap async functions to run in event loop
+        if asyncio.iscoroutinefunction(func):
+
+            def async_wrapper() -> None:
+                # Create new event loop for the background thread
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                try:
+                    loop.run_until_complete(func())
+                finally:
+                    loop.close()
+
+            wrapped_func = async_wrapper
+        else:
+            wrapped_func = func
+
+        # Explicitly pass timezone to CronTrigger to ensure the trigger uses the intended timezone,
+        # even though the scheduler also has a timezone setting.
+        trigger = CronTrigger(
+            hour=hour,
+            minute=minute,
+            day_of_week=day_of_week,
+            day=day,
+            month=month,
+            timezone=self.timezone,
+        )
+
+        self.scheduler.add_job(
+            wrapped_func,
+            trigger=trigger,
+            id=job_id,
+            replace_existing=True,
+        )
+        self.logger.info(
+            f"Registered job '{job_id}' - schedule: {month}/{day} {hour}:{minute} (day_of_week={day_of_week}) in {self.timezone}"
+        )
+
+    def remove_job(self, job_id: str) -> None:
+        """Remove a job from the scheduler"""
+        if self.scheduler is None:
+            return
+        try:
+            self.scheduler.remove_job(job_id)
+            self.logger.info(f"Removed cron job '{job_id}'")
+        except Exception as e:
+            self.logger.warning(f"Failed to remove job '{job_id}': {e}")
+
+    def list_jobs(self) -> list[dict[str, Any]]:
+        """List all scheduled jobs"""
+        if self.scheduler is None:
+            return []
+        return [
+            {
+                "id": job.id,
+                "next_run": str(job.next_run_time),
+                "trigger": str(job.trigger),
+            }
+            for job in self.scheduler.get_jobs()
+        ]

backend/python/app/services/jobs/README.md

@@ -0,0 +1,116 @@
+# Scheduled Jobs
+
+This directory contains all scheduled tasks (cron jobs) for the Food4Kids platform.
+
+### Architecture
+
+```
+app/
+  services/
+    implementations/
+      scheduler_service.py    # Central scheduler management
+    jobs/
+      __init__.py            # init_jobs() - registers all jobs
+      driver_history_jobs.py # Example job implementation
+      README.md              # This file
+```
+
+### How It Works
+
+1. On application startup, `SchedulerService` is initialized and started
+2. `init_jobs()` is called, which imports all job modules and registers them with the scheduler
+3. APScheduler runs jobs in background threads according to their cron schedules
+4. Async job functions are automatically wrapped to run in their own event loops
+5. On application shutdown, the scheduler is gracefully stopped
+
+## Creating Your Own Cron Job
+
+Follow these steps to add a new scheduled job:
+
+### Step 1: Create a Job File
+
+Create a new file in this directory (e.g., `email_jobs.py`, `cleanup_jobs.py`):
+
+```python
+"""Your job description"""
+import logging
+from app.dependencies.services import get_logger
+from app.models import async_session_maker_instance
+
+
+async def your_job_function() -> None:
+    """Description of what this job does
+    """
+    logger = get_logger()
+    
+    if async_session_maker_instance is None:
+        logger.error("Database session maker not initialized")
+        return
+    
+    try:
+        async with async_session_maker_instance() as session:
+            # Your job logic here
+            await session.commit()
+            logger.info("Job completed successfully")
+    except Exception as e:
+        logger.error(f"Error in job: {e}", exc_info=True)
+        if async_session_maker_instance:
+            try:
+                async with async_session_maker_instance() as session:
+                    await session.rollback()
+            except Exception:
+                pass
+        raise
+```
+
+### Step 2: Register the Job
+
+Add your job to `__init__.py`:
+
+```python
+def init_jobs(scheduler_service) -> None:
+    from .driver_history_jobs import process_daily_driver_history
+    from .email_jobs import your_job_function  # Import your new job
+    
+    # Existing jobs...
+    scheduler_service.add_cron_job(
+        process_daily_driver_history,
+        job_id="daily_driver_history",
+        hour=23,
+        minute=59,
+    )
+    
+    # Register your new job
+    scheduler_service.add_cron_job(
+        your_job_function,
+        job_id="your_job_id",
+        hour=9,
+        minute=0,
+        day_of_week=0,
+    )
+```
+
+### Step 3: Cron Schedule Parameters
+
+The `add_cron_job()` method accepts these parameters:
+
+- `job_id` (required): Unique identifier for the job
+- `hour` (default: "*"): Hour (0-23) or "*" for every hour
+- `minute` (default: "*"): Minute (0-59) or "*" for every minute
+- `day_of_week` (default: "*"): Day of week (0-6, 0=Monday) or "*" for every day
+- `day` (default: "*"): Day of month (1-31) or "*" for every day
+- `month` (default: "*"): Month (1-12) or "*" for every month
+
+### Testing
+
+To test your job manually, you can call it directly:
+
+```python
+from app.services.jobs.your_job_file import your_job_function
+import asyncio
+
+asyncio.run(your_job_function())
+```
+
+You can also create a test endpoint in development to trigger jobs manually.
+

backend/python/app/services/jobs/__init__.py

@@ -0,0 +1,23 @@
+"""Scheduled jobs - follows same pattern as routers"""
+
+from app.services.implementations.scheduler_service import SchedulerService
+
+
+def init_jobs(scheduler_service: SchedulerService) -> None:
+    """Initialize all scheduled jobs - add new jobs here
+
+    This function follows the same pattern as app.routers.init_app().
+    To add a new scheduled job:
+    1. Create a new file in this directory (e.g., email_jobs.py)
+    2. Define your job function
+    3. Import and register it here
+    """
+    from .driver_history_jobs import process_daily_driver_history
+
+    # Driver history mark daily routes as complete
+    scheduler_service.add_cron_job(
+        process_daily_driver_history,
+        job_id="daily_driver_history",
+        hour=23,
+        minute=59,
+    )