Merge: from main

.gitignore

@@ -1,2 +1,228 @@
-test.json
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#   Usually these files are written by a python script from a template
+#   before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+# Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+# uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+# poetry.lock
+# poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+# pdm.lock
+# pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+# pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# Redis
+*.rdb
+*.aof
+*.pid
+
+# RabbitMQ
+mnesia/
+rabbitmq/
+rabbitmq-data/
+
+# ActiveMQ
+activemq-data/
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#   JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#   and can be added to the global gitignore or merged into this file.  For a more nuclear
+#   option (not recommended) you can uncomment the following to ignore the entire idea folder.
+# .idea/
+
+# Abstra
+#   Abstra is an AI-powered process automation framework.
+#   Ignore directories containing user credentials, local state, and settings.
+#   Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#   Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#   that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#   and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#   you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/
+
+# Streamlit
+.streamlit/secrets.toml
+
+playground/psql_driver.ipynb
+Backend/Time complexity cheatsheet.pdf
+test/1000-data-science-questions-answers.json
+
+test/test.ipynb
+vector_db
+.gitignore
+
 test.ipynb
+.gitignore
+test.json

Backend/.gitignore

@@ -1,218 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[codz]
-*$py.class
-
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#   Usually these files are written by a python script from a template
-#   before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py.cover
-.hypothesis/
-.pytest_cache/
-cover/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-.pybuilder/
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-#   For a library or package, you might want to ignore these files since the code is
-#   intended to run in multiple environments; otherwise, check them in:
-# .python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-# Pipfile.lock
-
-# UV
-#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-# uv.lock
-
-# poetry
-#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
-# poetry.lock
-# poetry.toml
-
-# pdm
-#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
-#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
-#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
-# pdm.lock
-# pdm.toml
-.pdm-python
-.pdm-build/
-
-# pixi
-#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
-# pixi.lock
-#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
-#   in the .venv directory. It is recommended not to include this directory in version control.
-.pixi
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# Redis
-*.rdb
-*.aof
-*.pid
-
-# RabbitMQ
-mnesia/
-rabbitmq/
-rabbitmq-data/
-
-# ActiveMQ
-activemq-data/
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.envrc
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
-# pytype static type analyzer
-.pytype/
-
-# Cython debug symbols
-cython_debug/
-
-# PyCharm
-#   JetBrains specific template is maintained in a separate JetBrains.gitignore that can
-#   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#   and can be added to the global gitignore or merged into this file.  For a more nuclear
-#   option (not recommended) you can uncomment the following to ignore the entire idea folder.
-# .idea/
-
-# Abstra
-#   Abstra is an AI-powered process automation framework.
-#   Ignore directories containing user credentials, local state, and settings.
-#   Learn more at https://abstra.io/docs
-.abstra/
-
-# Visual Studio Code
-#   Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
-#   that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
-#   and can be added to the global gitignore or merged into this file. However, if you prefer, 
-#   you could uncomment the following to ignore the entire vscode folder
-# .vscode/
-
-# Ruff stuff:
-.ruff_cache/
-
-# PyPI configuration file
-.pypirc
-
-# Marimo
-marimo/_static/
-marimo/_lsp/
-__marimo__/
-
-# Streamlit
-.streamlit/secrets.toml
-
-playground/psql_driver.ipynb

Backend/app/api/deps.py

@@ -6,6 +6,9 @@ from jose import JWTError, jwt
 from app.database import async_session_maker
 from app.models import User
 from app.config import settings
+from fastapi import Request
+from chromadb import AsyncHttpClient
+from chromadb.api.models.Collection import Collection
 
 security  = HTTPBearer()
 
@@ -48,3 +51,16 @@ async def get_current_user(
     
     return user
 
+
+
+async def get_chroma_client(request: Request) -> AsyncHttpClient:
+    client = getattr(request.app.state, "chroma_client", None)
+    if client is None:
+        raise RuntimeError("ChromaDB client is not initialized in App State.")
+    return client
+
+def get_chroma_collection(request: Request) -> Collection:
+    collection = getattr(request.app.state, "chroma_collection", None)
+    if collection is None:
+        raise RuntimeError("ChromaDB Collection not loaded during application startup.")
+    return collection

Backend/app/api/v1/api.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter
-from app.api.v1.endpoints import auth, students
+from app.api.v1.endpoints import auth, students, quiz
 
 api_router = APIRouter()
 
@@ -16,3 +16,10 @@ api_router.include_router(
     prefix="/students",
     tags=["Students"]
 )
+
+# Include student routes
+api_router.include_router(
+    quiz.router,
+    prefix="/quiz",
+    tags=["quiz"]
+)

Backend/app/api/v1/endpoints/prompts.py

@@ -0,0 +1,47 @@
+SYSTEM_PROMPT = """
+You are an AI question-generation agent.
+Your task is to generate a batch of 10 high-quality MCQ questions strictly based on the following inputs:
+
+- {parsed_info}
+- {user_prompt}
+- {retrieved_docs}
+
+-----------------------
+GENERATION RULES
+-----------------------
+1. Generate exactly 10 MCQs.
+2. Use only information from the provided inputs.
+3. Each question must be unambiguous, factual, and supported by the given data.
+4. Each MCQ MUST have exactly four options.
+5. Only one correct answer is allowed.
+6. Explanations must be short and directly justify the answer.
+7. `User_response` must ALWAYS remain an empty string.
+8. Output MUST be a valid JSON array containing 10 objects.
+9. Output MUST contain nothing except the JSON array (no commentary or markdown).
+
+-----------------------
+REQUIRED JSON FORMAT FOR EACH QUESTION
+-----------------------
+{{
+    "question": "Which of the following CLI command can also be used to rename files?",
+    "options": [
+        "rm",
+        "mv",
+        "rm -r",
+        "none of the mentioned"
+    ],
+    "answer": "b",
+    "explanation": "mv stands for move.",
+    "User_response": ""
+}}
+
+-----------------------
+ANSWER KEY RULES
+-----------------------
+- 'a' -> options[0]
+- 'b' -> options[1]
+- 'c' -> options[2]
+- 'd' -> options[3]
+
+Strictly follow the JSON structure and generate exactly 10 MCQs.
+"""

Backend/app/api/v1/endpoints/quiz.py

@@ -0,0 +1,65 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from chromadb import AsyncHttpClient
+from app.models import User
+from app.api.deps import get_db, get_current_user, get_chroma_client
+from app.schema import Quiz_input, QuizOutput
+from .prompts import SYSTEM_PROMPT
+from fastapi import APIRouter, Depends, HTTPException
+from chromadb.api.models.Collection import Collection # Import Collection type
+from app.api.deps import get_chroma_collection
+from app.llm import call_llm
+router = APIRouter(prefix="/quiz")
+
+async def search_logic(query: str, collection: Collection):
+    results = await collection.query(
+        query_texts=[query],
+        n_results=5
+    )
+    return ''.join(results['documents'][0])
+
+@router.get("/search_docs")
+async def search_documents(
+    query: str,
+    collection: Collection = Depends(get_chroma_collection)
+):
+    try:
+        return await search_logic(query, collection)
+    except Exception as e:
+        raise HTTPException(500, f"ChromaDB Query Error: {e}")
+
+
+@router.post("/", response_model=QuizOutput, status_code=status.HTTP_201_CREATED)
+async def generate_quiz(
+    Input_model: Quiz_input, 
+    collection: Collection = Depends(get_chroma_collection), 
+    current_user: User = Depends(get_current_user)
+):
+    try:
+        query = Input_model.parsed_doc + Input_model.user_prompt
+        retrieved_context = await search_logic(query, collection)
+        
+
+        if not retrieved_context:
+            raise ValueError("No context available to generate quiz.")
+        prompt = await prompt_builder(Input_model.parsed_doc, Input_model.user_prompt, retrieved_context)
+        
+        quiz_data_obj = await call_llm(prompt)
+
+        return quiz_data_obj
+
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f'Invalid Input: {str(e)}'
+        )
+
+# #--------Helper Functions--------#
+
+
+async def prompt_builder(parsed_doc:str, user_prompt:str, docs:str=None):
+    prompt = SYSTEM_PROMPT.format(
+        parsed_info=parsed_doc,
+        user_prompt=user_prompt,
+        retrieved_docs=docs
+    )
+    return prompt

Backend/app/config.py

@@ -12,8 +12,14 @@ class Settings(BaseSettings):
 
     CORS_ORIGINS: list = ["*"]
 
+    chroma_host: str
+    chroma_port: int
+    chroma_collection: str
+
+    GROQ_API_KEY:str
+
     class Config:
         env_file = ".env"
-        extra = "ignore"  # optional
+        extra = "ignore"  # quiz
 
 settings = Settings()

Backend/app/llm.py

@@ -0,0 +1,35 @@
+import os
+from openai import OpenAI
+from pydantic import BaseModel, Field
+from typing import List, Optional, Any
+from app.schema.models import QuizOutput, QuizQuestion
+from app.config import settings
+
+client = OpenAI(
+    base_url="https://api.groq.com/openai/v1",
+    api_key=settings.GROQ_API_KEY
+)
+
+async def call_llm(prompt:str):
+    try:
+        response = client.chat.completions.create(
+            # CRUCIAL: Use the LiteLLM format: 'gemini/gemini-2.5-pro'
+            model="openai/gpt-oss-20b", 
+            messages=[
+                {"role": "user", "content": prompt}
+            ],
+            # Use the OpenAI parameter to request JSON output
+            response_format={"type": "json_object"}, 
+            temperature=0.7,
+        )
+
+        json_string = response.choices[0].message.content
+
+        import json
+        quiz_data = json.loads(json_string)
+        wrapped_data = {"quiz": quiz_data}
+        return QuizOutput.model_validate(wrapped_data)
+
+    except Exception as e:
+        print(f"Error calling LiteLLM/Gemini: {e}")
+        raise e

Backend/app/main.py

@@ -5,7 +5,11 @@ from datetime import datetime
 from app.config import settings
 from app.database import engine, Base
 from app.api.v1.api import api_router
+import chromadb
+from chromadb.api.models.Collection import Collection
+from dotenv import load_dotenv
 
+load_dotenv()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -16,6 +20,21 @@ async def lifespan(app: FastAPI):
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
     
+    client = await chromadb.AsyncHttpClient(
+        host=settings.chroma_host,
+        port=settings.chroma_port
+    )
+    app.state.chroma_client = client
+
+    try:
+        collection: Collection = await client.get_or_create_collection(settings.chroma_collection)
+        app.state.chroma_collection = collection
+
+        count = await collection.count()
+        print(f"Successfully loaded collection '{settings.chroma_collection}' with {count} documents.")
+    except Exception as e:
+        print(f"Failed to load ChromaDB collection: {e}")
+
     print("✅ Tables ready!")
     yield
     print("🧹 Server shutting down:", datetime.now())

Backend/app/models/tables.py

@@ -19,3 +19,4 @@ class User(Base):
     username: Mapped[str] =  mapped_column(String(50), unique=True, index=True)
     email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
     hashed_password: Mapped[str] = mapped_column(String(255))
+

Backend/app/schema/__init__.py

@@ -1,3 +1,3 @@
-from app.schema.models import StudentCreate, StudentUpdate, StudentResponse, UserCreate, Token, LoginRequest
+from app.schema.models import StudentCreate, StudentUpdate, StudentResponse, UserCreate, Token, LoginRequest, Quiz_input, QuizOutput
 
-__all__ = ["StudentCreate", "StudentUpdate", "StudentResponse", "UserCreate", "Token", "LoginRequest"]
+__all__ = ["StudentCreate", "StudentUpdate", "StudentResponse", "UserCreate", "Token", "LoginRequest", "Quiz_input", "QuizOutput"]

Backend/app/schema/models.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel, EmailStr, Field, field_validator, ConfigDict
-from typing import  Optional
+from typing import  Optional, Literal, List
 from datetime import datetime
 
 class StudentBase(BaseModel):
@@ -46,4 +46,19 @@ class Token(BaseModel):
     token_type: str
 
 class LoginResponse(Token):
-    username: str
+    username: str
+
+class Quiz_input(BaseModel):
+    parsed_doc: str
+    user_prompt: str
+    # choice: Literal["mcq", "code"]
+
+class QuizQuestion(BaseModel):
+    question: str 
+    options: List[str] = Field(..., min_items=2)
+    answer: str = Field(..., description="Correct answer key")
+    explanation: str
+    User_response: str = Field("", alias="User_response")
+
+class QuizOutput(BaseModel):
+    quiz: List[QuizQuestion] = Field(..., description="A list of 10 generated MCQ questions.")

Backend/app/vector_db.py

@@ -0,0 +1,4 @@
+import chromadb
+from chromadb.config import Settings
+
+chroma_client = None

Backend/requirements.txt

@@ -8,3 +8,4 @@ passlib[argon2]
 python-jose[cryptography]
 email-validator
 
+

RUN.md

@@ -0,0 +1,7 @@
+npm run dev (frontend)
+
+python run.py (Backend)
+
+sudo service postgresql start (POstgresql)
+
+chroma run --host 0.0.0.0 --port 8080 --path ./chroma_store (chroma_db)

test/db CRUD.py

@@ -1,88 +0,0 @@
-import asyncio
-from datetime import datetime
-from sqlalchemy import String, Integer, DateTime
-from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
-from sqlalchemy.orm  import DeclarativeBase, Mapped, mapped_column
-from sqlalchemy import select
-
-DATABASE_URL = "postgresql+asyncpg://postgres:690869@localhost:5432/studentdb"
-
-engine = create_async_engine(DATABASE_URL, echo=True)
-async_session = async_sessionmaker(engine, expire_on_commit=False)
-
-class Base(DeclarativeBase):
-    pass
-
-class Student(Base):
-    __tablename__ = "students"
-
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(100))
-    email: Mapped[str] = mapped_column(String(100))
-    age: Mapped[int]
-    grade: Mapped[str] = mapped_column(String(5))
-    created_at : Mapped[datetime] = mapped_column(default=datetime.utcnow)
-
-async def create_table():
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
-    print('Tables created successfully !')
-
-async def add_student(name:str, email: str, age: int, grade: str):
-    async with async_session() as session:
-        async with session.begin():
-            student = Student( name = name, email= email, age=age, grade=grade)
-            session.add(student)
-    print(f"added student{name}")
-
-async def list_students():
-    async with  async_session() as session:
-        result = await session.execute(select(Student))
-        students = result.scalars().all()
-
-        for s in students:
-            print(f" - {s.id}: {s.name}, {s.email}, Grade: {s.grade}, {s.age}")
-
-async def update_student(student_id: int, new_name: str = None, new_age : int = None, new_grade : int = None ):
-    async with async_session() as session:
-        async with session.begin():
-            student = await session.get(Student, student_id)
-            if not student:
-                print(f"❌ Student with id {student_id} not found.")
-                return
-            if new_name: 
-                student.name = new_name
-            if new_grade:
-                student.grade = new_grade
-            if new_age:
-                student.age = new_age
-
-            # await session.commit()  ## no need
-    print(f"✏️ Updated student ID {student_id}")
-
-async def delete_student(student_id: int):
-    async with async_session() as session:
-        async with session.begin():
-            student = await session.get(Student, student_id)
-            if not student:
-                print(f"❌ Student with id {student_id} not found.")
-                return
-            
-            await session.delete(student)
-            # await session.commit()  ## no need
-    print(f"🗑️ Deleted student ID {student_id}")
-
-
-async def main():
-    await create_table()
-    await add_student("Akshat Mehta", "akshat@example.com", 21, "A+")
-    await add_student("Nikhil Sharma", "nikhil@example.com", 22, "B")
-    await list_students()
-
-    await update_student(1, new_age=52)
-    await delete_student(2)
-    print('-'*100)
-    await list_students()
-
-if __name__ == "__main__":
-    asyncio.run(main())

test/db_fastapi_CRUD.py

@@ -1,127 +0,0 @@
-import asyncio
-from typing import List, Optional
-from contextlib import asynccontextmanager
-from datetime import datetime
-from fastapi import FastAPI, HTTPException, status
-from sqlalchemy import String, Integer, DateTime, select
-from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
-from sqlalchemy.orm  import DeclarativeBase, Mapped, mapped_column
-from pydantic import BaseModel, ConfigDict
-
-
-DATABASE_URL = "postgresql+asyncpg://postgres:690869@localhost:5432/studentdb"
-
-engine = create_async_engine(DATABASE_URL, echo=True)
-async_session = async_sessionmaker(engine, expire_on_commit=False)
-
-class Base(DeclarativeBase):
-    pass
-
-class Student(Base):
-    __tablename__ = "students"
-
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(100))
-    email: Mapped[str] = mapped_column(String(100))
-    age: Mapped[int]
-    grade: Mapped[str] = mapped_column(String(5))
-    created_at : Mapped[datetime] = mapped_column(default=datetime.utcnow)
-
-class StudentCreate(BaseModel):
-    name: str
-    email: str 
-    age: int
-    grade: str
-
-class StudentUpdate(BaseModel):
-    name: Optional[str] = None
-    email: Optional[str] = None
-    age: Optional[int] = None
-    grade: Optional[str] = None
-
-class StudentOut(BaseModel):
-    id: int
-    name: str
-    email: str 
-    age: int
-    grade: str
-    created_at: datetime
-
-    model_config = ConfigDict(from_attributes=True)
-
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
-        print("Tables created !!!!")
-    yield
-    await engine.dispose()
-    print("🔻 Database connection closed.")
-
-app = FastAPI(title="Async student CRUD API", lifespan=lifespan)
-
-@app.post("/students/", response_model=StudentOut, status_code=status.HTTP_201_CREATED)
-async def create_student(student: StudentCreate):
-    async with async_session() as session:
-        async with session.begin():
-            student = Student(**student.model_dump())
-            session.add(student)
-        await session.refresh(student)
-        return student
-
-@app.get("/students/", response_model=List[StudentOut])
-async def list_students():
-    async with  async_session() as session:
-        result = await session.execute(select(Student))
-        students = result.scalars().all()
-        return students
-
-@app.get("/students/{id}", response_model=StudentOut)
-async def get_student(id: int):
-    async with async_session() as session:
-        student = await session.get(Student, id)
-        if not student:
-            raise HTTPException(status_code=404, detail="Student not found")
-        return student
-
-@app.put("/students/{id}/", response_model=StudentOut)
-async def update_student(id: int, update_data: StudentUpdate):
-    async with async_session() as session:
-        async with session.begin():
-            student = await session.get(Student, id)
-            if not student:
-                raise HTTPException(status_code=404, detail="student not found")
-            
-            update_dict = update_data.model_dump(exclude_unset=True)
-            for key , value in update_dict.items():
-                setattr(student, key , value)
-        
-        await session.refresh(student)
-        return student
-
-@app.delete("/students/{id}", status_code=status.HTTP_204_NO_CONTENT)
-async def delete_student(id: int):
-    async with async_session() as session:
-        async with session.begin():
-            student = await session.get(Student, id)
-            if not student:
-                raise HTTPException(status_code=404, detail="student not found")
-            await session.delete(student)
-    return None
-
-
-# async def main():
-#     # await create_table()
-#     await add_student("Akshat Mehta", "akshat@example.com", 21, "A+")
-#     await add_student("Nikhil Sharma", "nikhil@example.com", 22, "B")
-#     await list_students()
-
-#     await update_student(1, new_age=52)
-#     await delete_student(2)
-#     print('-'*100)
-#     await list_students()
-
-# if __name__ == "__main__":
-#     asyncio.run(main())

test/main.py

@@ -1,445 +0,0 @@
-"""
-FastAPI + PostgreSQL Student Management System
-Complete async implementation with SQLAlchemy, authentication, and validation
-
-Requirements:
-pip install fastapi uvicorn sqlalchemy asyncpg psycopg2-binary python-jose[cryptography] passlib[bcrypt] python-multipart
-"""
-
-from fastapi import FastAPI, HTTPException, Depends, status
-from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-from fastapi.middleware.cors import CORSMiddleware
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
-from sqlalchemy import select, String
-from pydantic import BaseModel, EmailStr, Field, field_validator,  ConfigDict
-from typing import Optional, List
-from datetime import datetime, timedelta
-from jose import JWTError, jwt
-from passlib.context import CryptContext
-import os
-from contextlib import asynccontextmanager
-# ======================== CONFIGURATION ========================
-DATABASE_URL = os.getenv(
-    "DATABASE_URL", 
-    "postgresql+asyncpg://postgres:690869@172.26.157.164:5432/studentdb"
-)
-SECRET_KEY = os.getenv("SECRET_KEY", "production")
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 30
-
-# ======================== DATABASE SETUP ========================
-engine = create_async_engine(DATABASE_URL, echo=True)
-async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
-
-class Base(DeclarativeBase):
-    pass
-
-class Student(Base):
-    __tablename__ = "students"
-    
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    name: Mapped[str] = mapped_column(String(100))
-    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
-    age: Mapped[int]
-    grade: Mapped[str] = mapped_column(String(5))
-    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
-
-class User(Base):
-    __tablename__ = "users"
-    
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
-    hashed_password: Mapped[str] = mapped_column(String(255))
-
-# ======================== PYDANTIC MODELS ========================
-class StudentBase(BaseModel):
-    name: str = Field(..., min_length=2, max_length=100, description="Student full name")
-    email: EmailStr = Field(..., description="Student email address")
-    age: int = Field(..., ge=5, le=100, description="Student age (5-100)")
-    grade: str = Field(..., pattern="^[A-F][+-]?$", description="Grade (A-F with optional + or -)")
-    
-    @field_validator('name')
-    def validate_name(cls, v):
-        if not v.strip():
-            raise ValueError('Name cannot be empty or just whitespace')
-        return v.strip()
-    
-    @field_validator('grade')
-    def validate_grade(cls, v):
-        v = v.upper()
-        if v not in ['A+', 'A', 'A-', 'B+', 'B', 'B-', 'C+', 'C', 'C-', 'D+', 'D', 'D-', 'F']:
-            raise ValueError('Invalid grade format')
-        return v
-
-class StudentCreate(StudentBase):
-    pass
-
-class StudentUpdate(BaseModel):
-    name: Optional[str] = Field(None, min_length=2, max_length=100)
-    email: Optional[EmailStr] = None
-    age: Optional[int] = Field(None, ge=5, le=100)
-    grade: Optional[str] = Field(None, pattern="^[A-F][+-]?$")
-    
-    @field_validator('grade')
-    def validate_grade(cls, v):
-        if v is not None:
-            v = v.upper()
-            if v not in ['A+', 'A', 'A-', 'B+', 'B', 'B-', 'C+', 'C', 'C-', 'D+', 'D', 'D-', 'F']:
-                raise ValueError('Invalid grade format')
-        return v
-
-class StudentResponse(StudentBase):
-    id: int
-    created_at: datetime
-    
-    model_config = ConfigDict(from_attributes=True)
-
-class UserCreate(BaseModel):
-    username: str = Field(..., min_length=3, max_length=50)
-    password: str = Field(..., min_length=6, max_length=72)
-    
-    @field_validator('password')
-    def validate_password(cls, v):
-        if len(v.encode('utf-8')) > 72:
-            raise ValueError('Password cannot exceed 72 bytes')
-        return v
-
-class Token(BaseModel):
-    access_token: str
-    token_type: str
-
-# ======================== SECURITY ========================
-pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
-security = HTTPBearer()
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    # Truncate password to 72 bytes for bcrypt compatibility
-    password_bytes = plain_password.encode('utf-8')[:72]
-    plain_password_truncated = password_bytes.decode('utf-8', errors='ignore')
-    return pwd_context.verify(plain_password_truncated, hashed_password)
-
-def get_password_hash(password: str) -> str:
-    # Truncate password to 72 bytes for bcrypt compatibility
-    password_bytes = password.encode('utf-8')[:72]
-    password_truncated = password_bytes.decode('utf-8', errors='ignore')
-    return pwd_context.hash(password_truncated)
-
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    to_encode = data.copy()
-    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
-
-async def get_current_user(
-    credentials: HTTPAuthorizationCredentials = Depends(security),
-    db: AsyncSession = Depends(lambda: async_session_maker())
-):
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate credentials",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-    try:
-        token = credentials.credentials
-        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        username: str = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-    except JWTError:
-        raise credentials_exception
-    
-    result = await db.execute(select(User).filter(User.username == username))
-    user = result.scalar_one_or_none()
-    if user is None:
-        raise credentials_exception
-    return user
-
-# ======================== DATABASE DEPENDENCY ========================
-async def get_db():
-    async with async_session_maker() as session:
-        try:
-            yield session
-            await session.commit()
-        except Exception:
-            await session.rollback()
-            raise
-        finally:
-            await session.close()
-
-# new on event alternative 
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    print("🏗️ Server starting:", datetime.now())
-    print("🔧 Creating tables if they don't exist...")
-
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)        # Create tables
-
-    print("✅ Tables ready!")
-    yield
-    print("🧹 Server shutting down:", datetime.now())
-
-# ======================== FASTAPI APP ========================
-app = FastAPI(
-    title="Student Management API",
-    description="FastAPI + PostgreSQL with SQLAlchemy async",
-    version="1.0.0",
-    lifespan=lifespan
-
-)
-
-# CORS Configuration
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # In production, specify allowed origins
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-# ======================== AUTHENTICATION ROUTES ========================
-@app.post("/auth/register", response_model=dict, tags=["Authentication"])
-async def register(user: UserCreate, db: AsyncSession = Depends(get_db)):
-    """Register a new user"""
-    try:
-        result = await db.execute(select(User).filter(User.username == user.username))
-        existing_user = result.scalar_one_or_none()
-        
-        if existing_user:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Username already registered"
-            )
-        
-        new_user = User(
-            username=user.username,
-            hashed_password=get_password_hash(user.password)
-        )
-        db.add(new_user)
-        await db.commit()
-        
-        return {"message": "User registered successfully", "username": user.username}
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Registration failed: {str(e)}"
-        )
-
-@app.post("/auth/login", response_model=Token, tags=["Authentication"])
-async def login(username: str, password: str, db: AsyncSession = Depends(get_db)):
-    """Login and get access token"""
-    try:
-        result = await db.execute(select(User).filter(User.username == username))
-        user = result.scalar_one_or_none()
-        
-        if not user or not verify_password(password, user.hashed_password):
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Incorrect username or password",
-                headers={"WWW-Authenticate": "Bearer"},
-            )
-        
-        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-        access_token = create_access_token(
-            data={"sub": user.username},
-            expires_delta=access_token_expires
-        )
-        
-        return {"access_token": access_token, "token_type": "bearer"}
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Login failed: {str(e)}"
-        )
-
-# ======================== STUDENT CRUD ROUTES ========================
-@app.post("/students/", response_model=StudentResponse, status_code=status.HTTP_201_CREATED, tags=["Students"])
-async def create_student(
-    student: StudentCreate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Create a new student (Protected)"""
-    try:
-        # Check if email already exists
-        result = await db.execute(select(Student).filter(Student.email == student.email))
-        existing_student = result.scalar_one_or_none()
-        
-        if existing_student:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"Student with email {student.email} already exists"
-            )
-        
-        new_student = Student(**student.model_dump())
-        db.add(new_student)
-        await db.commit()
-        await db.refresh(new_student)
-        
-        return new_student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to create student: {str(e)}"
-        )
-
-@app.get("/students/", response_model=List[StudentResponse], tags=["Students"])
-async def get_all_students(
-    skip: int = 0,
-    limit: int = 100,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Get all students with pagination (Protected)"""
-    try:
-        result = await db.execute(
-            select(Student).offset(skip).limit(limit)
-        )
-        students = result.scalars().all()
-        return students
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to fetch students: {str(e)}"
-        )
-
-@app.get("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def get_student(
-    student_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Get a specific student by ID (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        return student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to fetch student: {str(e)}"
-        )
-
-@app.put("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def update_student(
-    student_id: int,
-    student_update: StudentUpdate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Update a student's information (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        # Update only provided fields
-        update_data = student_update.model_dump(exclude_unset=True)
-        
-        # Check email uniqueness if email is being updated
-        if "email" in update_data:
-            result = await db.execute(
-                select(Student).filter(
-                    Student.email == update_data["email"],
-                    Student.id != student_id
-                )
-            )
-            existing = result.scalar_one_or_none()
-            if existing:
-                raise HTTPException(
-                    status_code=status.HTTP_400_BAD_REQUEST,
-                    detail=f"Email {update_data['email']} is already in use"
-                )
-        
-        for key, value in update_data.items():
-            setattr(student, key, value)
-        
-        await db.commit()
-        await db.refresh(student)
-        
-        return student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to update student: {str(e)}"
-        )
-
-@app.patch("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def partial_update_student(
-    student_id: int,
-    student_update: StudentUpdate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Partially update a student (same as PUT for this implementation) (Protected)"""
-    return await update_student(student_id, student_update, db, current_user)
-
-@app.delete("/students/{student_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Students"])
-async def delete_student(
-    student_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Delete a student (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        await db.delete(student)
-        await db.commit()
-        
-        return None
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to delete student: {str(e)}"
-        )
-
-# ======================== HEALTH CHECK ========================
-@app.get("/", tags=["Health"])
-async def root():
-    """Health check endpoint"""
-    return {
-        "status": "healthy",
-        "message": "Student Management API is running",
-        "version": "1.0.0"
-    }
-
-# ======================== RUN APPLICATION ========================
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

test/test.ipynb

@@ -1,78 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "code",
-   "execution_count": 1,
-   "id": "d0e4e192",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import asyncio\n",
-    "from datetime import datetime\n",
-    "from sqlalchemy import String, Integer, DateTime\n",
-    "from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession\n",
-    "from sqlalchemy.orm  import DeclarativeBase, Mapped, mapped_column\n",
-    "from sqlalchemy import select\n",
-    "\n",
-    "DATABASE_URL = \"postgresql+asyncpg://postgres:690869@localhost:5432/studentdb\"\n",
-    "\n",
-    "engine = create_async_engine(DATABASE_URL, echo=True)\n",
-    "async_session = async_sessionmaker(engine, expire_on_commit=False)\n",
-    "\n",
-    "# class Base(DeclarativeBase):\n",
-    "#     pass\n",
-    "Base = DeclarativeBase()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": 4,
-   "id": "f2813d66",
-   "metadata": {},
-   "outputs": [
-    {
-     "data": {
-      "text/plain": [
-       "<sqlalchemy.orm.decl_api.DeclarativeBase at 0x7df42272b620>"
-      ]
-     },
-     "execution_count": 4,
-     "metadata": {},
-     "output_type": "execute_result"
-    }
-   ],
-   "source": [
-    "Base\n"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "52154570",
-   "metadata": {},
-   "outputs": [],
-   "source": []
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "prep",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.13.9"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}

test/test.py

@@ -1,445 +0,0 @@
-"""
-FastAPI + PostgreSQL Student Management System
-Complete async implementation with SQLAlchemy, authentication, and validation
-
-Requirements:
-pip install fastapi uvicorn sqlalchemy asyncpg psycopg2-binary python-jose[cryptography] passlib[bcrypt] python-multipart
-"""
-
-from fastapi import FastAPI, HTTPException, Depends, status
-from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-from fastapi.middleware.cors import CORSMiddleware
-from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
-from sqlalchemy import select, String
-from pydantic import BaseModel, EmailStr, Field, field_validator,  ConfigDict
-from typing import Optional, List
-from datetime import datetime, timedelta
-from jose import JWTError, jwt
-from passlib.context import CryptContext
-import os
-from contextlib import asynccontextmanager
-# ======================== CONFIGURATION ========================
-DATABASE_URL = os.getenv(
-    "DATABASE_URL", 
-    "postgresql+asyncpg://postgres:690869@172.26.157.164:5432/studentdb"
-)
-SECRET_KEY = os.getenv("SECRET_KEY", "production")
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 30
-
-# ======================== DATABASE SETUP ========================
-engine = create_async_engine(DATABASE_URL, echo=True)
-async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
-
-class Base(DeclarativeBase):
-    pass
-
-class Student(Base):
-    __tablename__ = "students"
-    
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    name: Mapped[str] = mapped_column(String(100))
-    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
-    age: Mapped[int]
-    grade: Mapped[str] = mapped_column(String(5))
-    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
-
-class User(Base):
-    __tablename__ = "users"
-    
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
-    hashed_password: Mapped[str] = mapped_column(String(255))
-
-# ======================== PYDANTIC MODELS ========================
-class StudentBase(BaseModel):
-    name: str = Field(..., min_length=2, max_length=100, description="Student full name")
-    email: EmailStr = Field(..., description="Student email address")
-    age: int = Field(..., ge=5, le=100, description="Student age (5-100)")
-    grade: str = Field(..., pattern="^[A-F][+-]?$", description="Grade (A-F with optional + or -)")
-    
-    @field_validator('name')
-    def validate_name(cls, v):
-        if not v.strip():
-            raise ValueError('Name cannot be empty or just whitespace')
-        return v.strip()
-    
-    @field_validator('grade')
-    def validate_grade(cls, v):
-        v = v.upper()
-        if v not in ['A+', 'A', 'A-', 'B+', 'B', 'B-', 'C+', 'C', 'C-', 'D+', 'D', 'D-', 'F']:
-            raise ValueError('Invalid grade format')
-        return v
-
-class StudentCreate(StudentBase):
-    pass
-
-class StudentUpdate(BaseModel):
-    name: Optional[str] = Field(None, min_length=2, max_length=100)
-    email: Optional[EmailStr] = None
-    age: Optional[int] = Field(None, ge=5, le=100)
-    grade: Optional[str] = Field(None, pattern="^[A-F][+-]?$")
-    
-    @field_validator('grade')
-    def validate_grade(cls, v):
-        if v is not None:
-            v = v.upper()
-            if v not in ['A+', 'A', 'A-', 'B+', 'B', 'B-', 'C+', 'C', 'C-', 'D+', 'D', 'D-', 'F']:
-                raise ValueError('Invalid grade format')
-        return v
-
-class StudentResponse(StudentBase):
-    id: int
-    created_at: datetime
-    
-    model_config = ConfigDict(from_attributes=True)
-
-class UserCreate(BaseModel):
-    username: str = Field(..., min_length=3, max_length=50)
-    password: str = Field(..., min_length=6, max_length=72)
-    
-    @field_validator('password')
-    def validate_password(cls, v):
-        if len(v.encode('utf-8')) > 72:
-            raise ValueError('Password cannot exceed 72 bytes')
-        return v
-
-class Token(BaseModel):
-    access_token: str
-    token_type: str
-
-# ======================== SECURITY ========================
-pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
-security = HTTPBearer()
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    # Truncate password to 72 bytes for bcrypt compatibility
-    password_bytes = plain_password.encode('utf-8')[:72]
-    plain_password_truncated = password_bytes.decode('utf-8', errors='ignore')
-    return pwd_context.verify(plain_password_truncated, hashed_password)
-
-def get_password_hash(password: str) -> str:
-    # Truncate password to 72 bytes for bcrypt compatibility
-    password_bytes = password.encode('utf-8')[:72]
-    password_truncated = password_bytes.decode('utf-8', errors='ignore')
-    return pwd_context.hash(password_truncated)
-
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    to_encode = data.copy()
-    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
-
-async def get_current_user(
-    credentials: HTTPAuthorizationCredentials = Depends(security),
-    db: AsyncSession = Depends(lambda: async_session_maker())
-):
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate credentials",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-    try:
-        token = credentials.credentials
-        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        username: str = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-    except JWTError:
-        raise credentials_exception
-    
-    result = await db.execute(select(User).filter(User.username == username))
-    user = result.scalar_one_or_none()
-    if user is None:
-        raise credentials_exception
-    return user
-
-# ======================== DATABASE DEPENDENCY ========================
-async def get_db():
-    async with async_session_maker() as session:
-        try:
-            yield session
-            await session.commit()
-        except Exception:
-            await session.rollback()
-            raise
-        finally:
-            await session.close()
-
-# new on event alternative 
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    print("🏗️ Server starting:", datetime.now())
-    print("🔧 Creating tables if they don't exist...")
-
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)        # Create tables
-
-    print("✅ Tables ready!")
-    yield
-    print("🧹 Server shutting down:", datetime.now())
-
-# ======================== FASTAPI APP ========================
-app = FastAPI(
-    title="Student Management API",
-    description="FastAPI + PostgreSQL with SQLAlchemy async",
-    version="1.0.0",
-    lifespan=lifespan
-
-)
-
-# CORS Configuration
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # In production, specify allowed origins
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-# ======================== AUTHENTICATION ROUTES ========================
-@app.post("/auth/register", response_model=dict, tags=["Authentication"])
-async def register(user: UserCreate, db: AsyncSession = Depends(get_db)):
-    """Register a new user"""
-    try:
-        result = await db.execute(select(User).filter(User.username == user.username))
-        existing_user = result.scalar_one_or_none()
-        
-        if existing_user:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Username already registered"
-            )
-        
-        new_user = User(
-            username=user.username,
-            hashed_password=get_password_hash(user.password)
-        )
-        db.add(new_user)
-        await db.commit()
-        
-        return {"message": "User registered successfully", "username": user.username}
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Registration failed: {str(e)}"
-        )
-
-@app.post("/auth/login", response_model=Token, tags=["Authentication"])
-async def login(username: str, password: str, db: AsyncSession = Depends(get_db)):
-    """Login and get access token"""
-    try:
-        result = await db.execute(select(User).filter(User.username == username))
-        user = result.scalar_one_or_none()
-        
-        if not user or not verify_password(password, user.hashed_password):
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Incorrect username or password",
-                headers={"WWW-Authenticate": "Bearer"},
-            )
-        
-        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-        access_token = create_access_token(
-            data={"sub": user.username},
-            expires_delta=access_token_expires
-        )
-        
-        return {"access_token": access_token, "token_type": "bearer"}
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Login failed: {str(e)}"
-        )
-
-# ======================== STUDENT CRUD ROUTES ========================
-@app.post("/students/", response_model=StudentResponse, status_code=status.HTTP_201_CREATED, tags=["Students"])
-async def create_student(
-    student: StudentCreate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Create a new student (Protected)"""
-    try:
-        # Check if email already exists
-        result = await db.execute(select(Student).filter(Student.email == student.email))
-        existing_student = result.scalar_one_or_none()
-        
-        if existing_student:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"Student with email {student.email} already exists"
-            )
-        
-        new_student = Student(**student.model_dump())
-        db.add(new_student)
-        await db.commit()
-        await db.refresh(new_student)
-        
-        return new_student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to create student: {str(e)}"
-        )
-
-@app.get("/students/", response_model=List[StudentResponse], tags=["Students"])
-async def get_all_students(
-    skip: int = 0,
-    limit: int = 100,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Get all students with pagination (Protected)"""
-    try:
-        result = await db.execute(
-            select(Student).offset(skip).limit(limit)
-        )
-        students = result.scalars().all()
-        return students
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to fetch students: {str(e)}"
-        )
-
-@app.get("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def get_student(
-    student_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Get a specific student by ID (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        return student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to fetch student: {str(e)}"
-        )
-
-@app.put("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def update_student(
-    student_id: int,
-    student_update: StudentUpdate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Update a student's information (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        # Update only provided fields
-        update_data = student_update.model_dump(exclude_unset=True)
-        
-        # Check email uniqueness if email is being updated
-        if "email" in update_data:
-            result = await db.execute(
-                select(Student).filter(
-                    Student.email == update_data["email"],
-                    Student.id != student_id
-                )
-            )
-            existing = result.scalar_one_or_none()
-            if existing:
-                raise HTTPException(
-                    status_code=status.HTTP_400_BAD_REQUEST,
-                    detail=f"Email {update_data['email']} is already in use"
-                )
-        
-        for key, value in update_data.items():
-            setattr(student, key, value)
-        
-        await db.commit()
-        await db.refresh(student)
-        
-        return student
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to update student: {str(e)}"
-        )
-
-@app.patch("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
-async def partial_update_student(
-    student_id: int,
-    student_update: StudentUpdate,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Partially update a student (same as PUT for this implementation) (Protected)"""
-    return await update_student(student_id, student_update, db, current_user)
-
-@app.delete("/students/{student_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Students"])
-async def delete_student(
-    student_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(get_current_user)
-):
-    """Delete a student (Protected)"""
-    try:
-        result = await db.execute(select(Student).filter(Student.id == student_id))
-        student = result.scalar_one_or_none()
-        
-        if not student:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Student with ID {student_id} not found"
-            )
-        
-        await db.delete(student)
-        await db.commit()
-        
-        return None
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to delete student: {str(e)}"
-        )
-
-# ======================== HEALTH CHECK ========================
-@app.get("/", tags=["Health"])
-async def root():
-    """Health check endpoint"""
-    return {
-        "status": "healthy",
-        "message": "Student Management API is running",
-        "version": "1.0.0"
-    }
-
-# ======================== RUN APPLICATION ========================
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

test/voice.py

@@ -1,62 +0,0 @@
-import sounddevice as sd
-import numpy as np
-from faster_whisper import WhisperModel
-import ollama
-import subprocess
-import time
-
-# --- Configuration ---
-# STT Model: 'tiny' or 'base' are fast; 'small' is accurate.
-stt_model = WhisperModel("base.en", device="cuda", compute_type="float16") 
-
-def record_audio(duration=5, samplerate=16000):
-    """Simple recorder - in production, use VAD (Voice Activity Detection) to stop automatically."""
-    print("Listening...")
-    recording = sd.rec(int(duration * samplerate), samplerate=samplerate, channels=1, dtype='float32')
-    sd.wait()
-    return recording.flatten()
-
-def transcribe(audio_data):
-    """Convert Audio to Text"""
-    segments, info = stt_model.transcribe(audio_data, beam_size=5)
-    text = " ".join([segment.text for segment in segments])
-    return text.strip()
-
-def generate_response(prompt):
-    """Send text to Ollama (The Brain)"""
-    response = ollama.chat(model='llama3.2', messages=[
-      {'role': 'user', 'content': prompt},
-    ])
-    return response['message']['content']
-
-def speak(text):
-    """Convert Text to Audio (The Mouth) using Piper"""
-    # Assuming you have the piper binary and a voice model downloaded
-    # Command line: echo "text" | piper --model en_US-lessac-medium.onnx --output_file output.wav
-    
-    # For this example, we'll use a generic speak command for macOS/Linux 
-    # (Replace with Piper or pyttsx3 for true local/cross-platform)
-    subprocess.call(["say", text]) 
-
-# --- Main Loop ---
-def main():
-    print("Voice Agent Started (Ctrl+C to stop)")
-    while True:
-        # 1. Listen
-        audio_data = record_audio(duration=4) # Fixed duration for simplicity
-        
-        # 2. Transcribe
-        user_text = transcribe(audio_data)
-        if not user_text: continue
-        
-        print(f"You: {user_text}")
-        
-        # 3. Think
-        ai_response = generate_response(user_text)
-        print(f"AI: {ai_response}")
-        
-        # 4. Speak
-        speak(ai_response)
-
-if __name__ == "__main__":
-    main()