feat: added fastapi-mongo sample

README.md

@@ -27,6 +27,8 @@ This repo contains the samples for [Keploy's](https://keploy.io) integration wit
 
 5. [Flask-Redis](https://github.com/keploy/samples-python/tree/main/flask-redis) - This Flask-based application provides a book management system utilizing Redis for caching and storage. It supports adding, retrieving, updating, and deleting book records, with optimized search functionality and cache management for improved performance. The API endpoints ensure efficient data handling and quick access to book information.
 
+6. [FastAPI-Mongo](www.github.com/keploy/samples-python/tree/main/fastapi-mongo) - This application is for a Quiz Platform that generated Quizzes using LLMs. This project demonstrates how to use Keploy API Test Generator to automatically generate API tests for a FastAPI application that interacts with a MongoDB database.
+
 ## Community Support ❤️
 
 ### 🤔 Questions?

fastapi-mongo/.gitignore

@@ -0,0 +1,4 @@
+**/__pycache__/**
+**/__init__.py/**
+__init__.py
+.DS_Store

fastapi-mongo/Dockerfile

@@ -0,0 +1,32 @@
+# 1. Use a lightweight Python 3.10 image
+FROM python:3.10-slim
+
+# 2. Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+# 3. Set working directory
+WORKDIR /app
+
+# 4. Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    gcc \
+    libpq-dev \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# 5. Copy requirements.txt from the root folder
+COPY ../requirements.txt .
+
+# 6. Install Python dependencies
+RUN pip install --upgrade pip && pip install -r requirements.txt
+
+# 7. Copy everything from quizzly folder (backend code)
+COPY . .
+
+# 8. Expose port
+EXPOSE 8000
+
+# 9. Run FastAPI app with uvicorn
+CMD ["uvicorn", "quizzly.main:app", "--host", "0.0.0.0", "--port", "8000"]

fastapi-mongo/auth/jwt_handler.py

@@ -0,0 +1,20 @@
+from jose import jwt
+from datetime import datetime, timedelta
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+SECRET_KEY = os.getenv("SECRET_KEY")
+ALGORITHM = "HS256"
+
+
+def create_access_token(data: dict, expires_delta: timedelta = timedelta(days=1)):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + expires_delta
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+def decode_access_token(token: str):
+    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

fastapi-mongo/core/config.py

@@ -0,0 +1,17 @@
+# app/core/config.py
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    MONGODB_URI: str
+    SECRET_KEY: str
+    FRONTEND_URL: str
+    GEMINI_API_KEY: str
+    YOUTUBE_API_KEY: str
+    CORS_ORIGINS: str
+
+    class Config:
+        env_file = ".env"
+
+
+settings = Settings()

fastapi-mongo/db/mongodb.py

@@ -0,0 +1,6 @@
+# app/db/mongodb.py
+from motor.motor_asyncio import AsyncIOMotorClient
+from quizzly.core.config import settings
+
+client = AsyncIOMotorClient(settings.MONGODB_URI)
+db = client.quiz_app

fastapi-mongo/keploy-api-test-generator/README.md

@@ -0,0 +1,9 @@
+### Keploy API Test Generator for FastAPI with MongoDB
+
+This project demonstrates how to use Keploy to automatically generate API tests for a FastAPI application that interacts with a MongoDB database. Keploy captures the API requests and responses, allowing you to create comprehensive test cases effortlessly.
+
+Keploy API Test Generator: https://keploy.io/docs/running-keploy/api-test-generator/
+
+### About the Project
+
+This is FastAPI + MongoDB backend written for a Quiz Application. The application allows users to perform CRUD operations on quiz data stored in a MongoDB database.

fastapi-mongo/main.py

@@ -0,0 +1,28 @@
+# app/main.py
+from fastapi import FastAPI
+from quizzly.routes import parent, quiz, content
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.cors import CORSMiddleware
+from quizzly.routes import parent, quiz
+from quizzly.core.config import settings
+
+
+app = FastAPI()
+# Read and parse CORS origins
+origins_str = settings.CORS_ORIGINS
+origins = [origin.strip() for origin in origins_str.split(",") if origin.strip()]
+
+print(origins)
+
+# Enable CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(parent.router, prefix="/parent", tags=["Parent"])
+app.include_router(quiz.router, prefix="/quiz", tags=["Quiz"])
+app.include_router(content.router, prefix="/content", tags=["Content"])

fastapi-mongo/models/parent.py

@@ -0,0 +1,17 @@
+# app/models/parent.py
+from pydantic import BaseModel, Field
+from typing import Optional
+from bson import ObjectId
+
+
+class ParentCreate(BaseModel):
+    username: str
+    password: str
+    name: Optional[str]
+
+
+class Parent(BaseModel):
+    id: Optional[str] = Field(default_factory=lambda: str(ObjectId()), alias="_id")
+    name: str
+    username: str
+    password: str

fastapi-mongo/models/quiz.py

@@ -0,0 +1,82 @@
+# app/models/quiz.py
+
+from pydantic import BaseModel, Field, HttpUrl
+from typing import List, Optional
+from bson import ObjectId
+from datetime import datetime
+import pytz
+
+IST = pytz.timezone("Asia/Kolkata")
+
+
+class EndQuizRequest(BaseModel):
+    name: str
+
+
+class QuizStart(BaseModel):
+    password: int
+    name: str
+
+
+class AnswerSubmission(BaseModel):
+    name: str
+    answers: List[str]
+
+
+# Question model to represent individual question details
+class Question(BaseModel):
+    question: str
+    choice_A: str
+    choice_B: str
+    choice_C: str
+    choice_D: str
+    answer: str  # Should be one of "A", "B", "C", or "D"
+    is_correct: Optional[bool] = False
+
+
+# QuizCreate model – input from client
+class QuizCreate(BaseModel):
+    name: str
+    subject: str
+    num_questions: int
+    # trigger_link: HttpUrl
+    # questions: List[Question]
+    topic: str
+    difficulty_level: int
+
+
+class QuestionUpdate(BaseModel):
+    question: str
+    choice_A: str
+    choice_B: str
+    choice_C: str
+    choice_D: str
+    answer: str
+    is_correct: bool
+
+
+# Main Quiz model – includes fields auto-populated by backend
+class Quiz(BaseModel):
+    id: str = Field(default_factory=lambda: str(ObjectId()), alias="_id")
+    name: str
+    subject: str
+    created_by: str
+    created_at: datetime = Field(default_factory=datetime.now(IST))
+    metadata_fields: Optional[dict] = Field(default_factory=dict)
+    trigger_link: HttpUrl
+    taken_by: List[str]
+    num_questions: int
+    questions: List[Question]
+    user_responses: List[dict]
+    is_started: bool = False
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    exec_time: Optional[float] = None  # in seconds
+    topic: str
+    difficulty_level: int
+    password: int
+    is_executed: bool = False
+
+    class Config:
+        json_encoders = {ObjectId: str, datetime: lambda v: v.isoformat()}
+        populate_by_name = True

fastapi-mongo/routes/content.py

@@ -0,0 +1,73 @@
+from fastapi import APIRouter, HTTPException
+from typing import List
+import json
+from googleapiclient.discovery import build
+from duckduckgo_search import DDGS
+from quizzly.core.config import settings
+import os
+from pydantic import BaseModel
+
+router = APIRouter()
+
+class TopicsRequest(BaseModel):
+    topics: List[str]
+    num_results: int = 5
+
+@router.post("/articles")
+async def get_articles(request: TopicsRequest):
+    """
+    Get articles based on provided topics using DuckDuckGo search
+    """
+    try:
+        results = {}
+        with DDGS() as ddgs:
+            for topic in request.topics:
+                search_results = ddgs.text(topic, max_results=request.num_results)
+                articles = []
+                for res in search_results:
+                    title = res.get('title')
+                    url = res.get('href')
+                    if title and url:
+                        articles.append({'title': title, 'url': url})
+                results[topic] = articles
+        return results
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/youtube")
+async def get_youtube_videos(request: TopicsRequest):
+    """
+    Get YouTube videos with thumbnails based on provided topics
+    """
+    if not settings.YOUTUBE_API_KEY:
+        raise HTTPException(status_code=500, detail="YouTube API key not configured")
+
+    try:
+        youtube = build('youtube', 'v3', developerKey=settings.YOUTUBE_API_KEY)
+        results = {}
+
+        for topic in request.topics:
+            search_response = youtube.search().list(
+                q=topic,
+                part='snippet',
+                type='video',
+                maxResults=request.num_results
+            ).execute()
+
+            youtube_data = []
+            for item in search_response['items']:
+                video_title = item['snippet']['title']
+                video_url = f"https://www.youtube.com/watch?v={item['id']['videoId']}"
+                video_id = item['id']['videoId']
+                thumbnail_url = item['snippet']['thumbnails']['high']['url']
+                youtube_data.append({
+                    'title': video_title,
+                    'video_url': video_url,
+                    'thumbnail_url': thumbnail_url
+                })
+
+            results[topic] = youtube_data
+
+        return results
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

fastapi-mongo/routes/parent.py

@@ -0,0 +1,36 @@
+# app/routes/parent.py
+import bcrypt
+from fastapi import APIRouter, HTTPException
+from quizzly.models.parent import Parent, ParentCreate
+from quizzly.db.mongodb import db
+from quizzly.utils.jwt import create_token
+from pydantic import BaseModel
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+
+router = APIRouter()
+
+
+@router.post("/register")
+async def register_parent(data: ParentCreate):
+    parent_data = Parent(**data.dict())
+    parent_data.password = bcrypt.hashpw(parent_data.password.encode(), bcrypt.gensalt()).decode()
+    await db.parents.insert_one(parent_data.model_dump(by_alias=True))
+    parent = await db.parents.find_one({"username": parent_data.username})
+    token = create_token({"id": str(parent["_id"])})
+    return {"message": "Parent registered successfully", "access_token": token}
+
+
+@router.post("/login")
+async def login(data: LoginRequest):
+    parent = await db.parents.find_one({"username": data.username})
+    if not parent:
+        raise HTTPException(status_code=404, detail="User not found")
+    if not bcrypt.checkpw(data.password.encode(), parent["password"].encode()):
+        raise HTTPException(status_code=401, detail="Invalid password")
+    token = create_token({"id": str(parent["_id"])})
+    return {"access_token": token}