Question Scheduler

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.13-slim
+FROM python:3.11-slim
 
 WORKDIR /app
 

app/logging_config.py

@@ -0,0 +1,28 @@
+import logging
+import sys
+
+
+def setup_logging(service_name: str):
+    formatter = logging.Formatter(
+        fmt=f"%(asctime)s | {service_name} | %(levelname)s | %(name)s | %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(formatter)
+
+    root_logger = logging.getLogger()
+    root_logger.setLevel(logging.INFO)  # default
+    root_logger.handlers.clear()  # avoid duplicate logs
+    root_logger.addHandler(handler)
+
+    # Tune FastAPI/uvicorn loggers too
+    logging.getLogger("uvicorn").handlers.clear()
+    logging.getLogger("uvicorn.error").handlers.clear()
+    logging.getLogger("uvicorn.access").handlers.clear()
+
+    logging.getLogger("uvicorn").addHandler(handler)
+    logging.getLogger("uvicorn.error").addHandler(handler)
+    logging.getLogger("uvicorn.access").addHandler(handler)
+
+    return root_logger

app/main.py

@@ -1,19 +1,50 @@
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
 
 from app.api.main import api_router
+from app.logging_config import setup_logging
+from app.scheduler import shutdown_scheduler, start_scheduler
 from app.settings import settings
 
+logger = setup_logging("Generator-Service")
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    """Start up and shut down events for the FastAPI app."""
+    logger.info("Starting FastAPI app...")
+    start_scheduler()
+    try:
+        yield
+    finally:
+        shutdown_scheduler()
+
+
 app = FastAPI(
     title="Generator Service",
     description="API for generating and managing coding problems",
     version="0.1.0",
+    lifespan=lifespan,
 )
 
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+    """Handle all uncaught exceptions globally."""
+    logger.exception("Unhandled error %s in request %s", exc, request.url)
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "Internal server error"},
+    )
+
+
 app.include_router(api_router, prefix=settings.API_V1_STR, tags=["v1"])
 
 
 # Health check endpoint
 @app.get("/health")
 async def health_check():
+    """Health check endpoint."""
     return {"status": "healthy", "service": "JamAndFlow-generator"}

app/scheduler.py

@@ -0,0 +1,74 @@
+import json
+import logging
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from app.services.questions import (add_daily_question_to_mongodb,
+                                    generate_daily_question)
+from app.utils.generic_functions import execute_with_retries
+
+logger = logging.getLogger(__name__)
+
+
+class SchedulerManager:
+    """Manages the APScheduler instance and scheduled jobs."""
+
+    def __init__(self):
+        self.scheduler = AsyncIOScheduler()
+        self.job = None
+
+    # TODO/question: should we raise error if job fails after retries?
+    async def _generate_and_store_daily_question(self):
+        """Generate a daily question and store it in MongoDB."""
+        try:
+            question = generate_daily_question()
+            question_json = json.loads(question)
+            add_daily_question_to_mongodb(question_json)
+        except (json.JSONDecodeError, TypeError, RuntimeError) as e:
+            logger.error("Error running schedul daily job: %s", e)
+        except BaseException as e:
+            logger.error("Unexpected critical error: %s", e)
+
+    async def _daily_question_task(self):
+        """Wrapper to execute the daily question generation with retries."""
+        await execute_with_retries(
+            self._generate_and_store_daily_question, max_retries=3, delay=5
+        )
+    #TODO: need to test with 24 hour interval
+    def start(self, interval_hour: int):
+        """Start the scheduler and add the daily question job."""
+        try:
+            self.job = self.scheduler.add_job(
+                self._daily_question_task,
+                "interval",
+                hour=24,
+                id="daily_question_job",
+                replace_existing=True,
+            )
+            self.scheduler.start()
+            logger.info(
+                "Scheduler started successfully with interval %d hour.", interval_hour
+            )
+        except Exception as e:
+            logger.error("Error starting scheduler: %s", e)
+            raise e
+
+    def shutdown(self):
+        """Shutdown the scheduler."""
+        if self.scheduler.running:
+            self.scheduler.shutdown()
+        logger.info("Scheduler shutdown completed.")
+
+
+# Singleton instance for use in FastAPI app
+scheduler_manager = SchedulerManager()
+
+
+def start_scheduler():
+    """Start the scheduler with a 1-minute interval for testing purposes."""
+    scheduler_manager.start(interval_hour=24)
+
+
+def shutdown_scheduler():
+    """Shutdown the scheduler."""
+    scheduler_manager.shutdown()

app/services/questions.py

@@ -18,6 +18,7 @@ def generate_daily_question(user_prompt: str = None) -> any:
     return response.content
 
 
+# TODO: Add error handling
 def add_daily_question_to_mongodb(question: dict) -> None:
     """Add a daily question to the MongoDB collection."""
     question_document = {

app/utils/generic_functions.py

@@ -0,0 +1,32 @@
+import logging
+import time
+
+logger = logging.getLogger(__name__)
+
+
+def execute_with_retries(func, *args, max_retries=3, delay=2, **kwargs):
+    """Execute a function with retries on failure.
+
+    Args:
+        func (callable): The function to execute.
+        max_retries (int): Maximum number of retries.
+        delay (int): Delay in seconds between retries.
+        *args: Positional arguments to pass to the function.
+        **kwargs: Keyword arguments to pass to the function.
+
+    Returns:
+        The result of the function if successful.
+
+    Raises:
+        Exception: If all retries fail, the last exception is raised.
+    """
+    attempt = 0
+    while attempt < max_retries:
+        try:
+            return func(*args, **kwargs)
+        except (RuntimeError, ValueError, TypeError) as e:
+            attempt += 1
+            logger.warning("Attempt %d failed with error: %s", attempt, e)
+            time.sleep(delay)
+
+    logger.error("All retry attempts failed. for function {func.__name__}")

requirements.txt

@@ -6,8 +6,8 @@ alembic~=1.16.2
 psycopg2-binary~=2.9.10
 python-dotenv
 pydantic
-pydantic-settings
 pymongo~=4.5.0
+APScheduler==3.11.0
 chromadb
 
 # LangChain and Hugging Face
@@ -16,4 +16,4 @@ langchain-huggingface
 langchain-community
 langchain_chroma
 huggingface-hub
-sentence_transformers
+sentence_transformers