From 33df91f1f1ff2d2220baa5ae872aa9b9f523320b Mon Sep 17 00:00:00 2001 From: KevinRawal Date: Sat, 6 Sep 2025 20:20:16 +0530 Subject: [PATCH 1/3] added schedular to generate daily questions --- Dockerfile | 2 +- app/api/routes/questions.py | 11 ++-- app/logging_config.py | 28 +++++++++ app/main.py | 35 ++++++++++- app/pipelines/daily_question_pipeline.py | 8 ++- app/scheduler.py | 76 ++++++++++++++++++++++++ app/services/questions.py | 1 + app/utils/generic_functions.py | 32 ++++++++++ requirements.txt | 4 +- 9 files changed, 185 insertions(+), 12 deletions(-) create mode 100644 app/logging_config.py create mode 100644 app/scheduler.py create mode 100644 app/utils/generic_functions.py diff --git a/Dockerfile b/Dockerfile index e0598e9..396b425 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.13-slim +FROM python:3.11-slim WORKDIR /app diff --git a/app/api/routes/questions.py b/app/api/routes/questions.py index e91e5b6..cbba7d1 100644 --- a/app/api/routes/questions.py +++ b/app/api/routes/questions.py @@ -3,10 +3,13 @@ from fastapi import APIRouter from app.schemas.utils import TechDescription, UserPrompt -from app.services.questions import (add_daily_question_to_mongodb, - add_tech_description_to_store) -from app.services.questions import \ - generate_daily_question as generate_daily_question_service +from app.services.questions import ( + add_daily_question_to_mongodb, + add_tech_description_to_store, +) +from app.services.questions import ( + generate_daily_question as generate_daily_question_service, +) from app.services.questions import get_most_recent_daily_question router = APIRouter() diff --git a/app/logging_config.py b/app/logging_config.py new file mode 100644 index 0000000..84bef78 --- /dev/null +++ b/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 diff --git a/app/main.py b/app/main.py index f6c2b4b..b6af06c 100644 --- a/app/main.py +++ b/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"} diff --git a/app/pipelines/daily_question_pipeline.py b/app/pipelines/daily_question_pipeline.py index 8ec1628..f328371 100644 --- a/app/pipelines/daily_question_pipeline.py +++ b/app/pipelines/daily_question_pipeline.py @@ -1,8 +1,10 @@ from typing import Dict -from langchain_core.prompts import (ChatPromptTemplate, - HumanMessagePromptTemplate, - SystemMessagePromptTemplate) +from langchain_core.prompts import ( + ChatPromptTemplate, + HumanMessagePromptTemplate, + SystemMessagePromptTemplate, +) from langchain_core.runnables import RunnableParallel from app.config.vectorestore import chroma_db diff --git a/app/scheduler.py b/app/scheduler.py new file mode 100644 index 0000000..5591f7a --- /dev/null +++ b/app/scheduler.py @@ -0,0 +1,76 @@ +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 + ) + + 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", + minutes=1, + 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() diff --git a/app/services/questions.py b/app/services/questions.py index 9344d25..7c5c744 100644 --- a/app/services/questions.py +++ b/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 = { diff --git a/app/utils/generic_functions.py b/app/utils/generic_functions.py new file mode 100644 index 0000000..48fc701 --- /dev/null +++ b/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__}") diff --git a/requirements.txt b/requirements.txt index c9ac474..a0c985e 100644 --- a/requirements.txt +++ b/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 \ No newline at end of file +sentence_transformers From 3fe21ed04dc6ed4ee797f0a46462e3dde24e0830 Mon Sep 17 00:00:00 2001 From: KevinRawal Date: Sat, 6 Sep 2025 20:21:14 +0530 Subject: [PATCH 2/3] minor changes --- app/scheduler.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index 5591f7a..304c3d2 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -36,14 +36,14 @@ async def _daily_question_task(self): 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", - minutes=1, + hour=24, id="daily_question_job", replace_existing=True, ) From a66a1a5703eff91dd7e60ea765425491e544a4f0 Mon Sep 17 00:00:00 2001 From: KevinRawal Date: Sat, 6 Sep 2025 20:38:29 +0530 Subject: [PATCH 3/3] lint fix --- app/api/routes/questions.py | 11 ++++------- app/pipelines/daily_question_pipeline.py | 8 +++----- app/scheduler.py | 6 ++---- 3 files changed, 9 insertions(+), 16 deletions(-) diff --git a/app/api/routes/questions.py b/app/api/routes/questions.py index cbba7d1..e91e5b6 100644 --- a/app/api/routes/questions.py +++ b/app/api/routes/questions.py @@ -3,13 +3,10 @@ from fastapi import APIRouter from app.schemas.utils import TechDescription, UserPrompt -from app.services.questions import ( - add_daily_question_to_mongodb, - add_tech_description_to_store, -) -from app.services.questions import ( - generate_daily_question as generate_daily_question_service, -) +from app.services.questions import (add_daily_question_to_mongodb, + add_tech_description_to_store) +from app.services.questions import \ + generate_daily_question as generate_daily_question_service from app.services.questions import get_most_recent_daily_question router = APIRouter() diff --git a/app/pipelines/daily_question_pipeline.py b/app/pipelines/daily_question_pipeline.py index f328371..8ec1628 100644 --- a/app/pipelines/daily_question_pipeline.py +++ b/app/pipelines/daily_question_pipeline.py @@ -1,10 +1,8 @@ from typing import Dict -from langchain_core.prompts import ( - ChatPromptTemplate, - HumanMessagePromptTemplate, - SystemMessagePromptTemplate, -) +from langchain_core.prompts import (ChatPromptTemplate, + HumanMessagePromptTemplate, + SystemMessagePromptTemplate) from langchain_core.runnables import RunnableParallel from app.config.vectorestore import chroma_db diff --git a/app/scheduler.py b/app/scheduler.py index 304c3d2..b62c645 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -3,10 +3,8 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler -from app.services.questions import ( - add_daily_question_to_mongodb, - generate_daily_question, -) +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__)