WIP: Making sure the clearance system can run live with Fastapi

.gitignore

@@ -0,0 +1,174 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$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
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.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/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc

README copy.md

@@ -0,0 +1 @@
+# clearance_stud

main.py

@@ -0,0 +1,191 @@
+from fastapi import FastAPI, HTTPException, Depends, status
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from typing import List # Import List for type hinting
+from src.database import connect_db, disconnect_db, database # Also import database instance if needed directly in main
+from src.models import (
+    StudentCreate, Student, ClearanceStatusCreate,
+    ClearanceStatus, ClearanceDetail, DeviceRegister,
+    DeviceResponse, TagScan
+) # Pydantic models
+from src import crud # Database operations (now uses databases)
+from src.auth import verify_api_key # API key verification dependency (now uses databases)
+
+# FastAPI app instance
+app = FastAPI(title="Undergraduate Clearance System API", version="1.0.0")
+
+# Enable CORS (Cross-Origin Resource Sharing)
+# This allows your frontend (running on a different origin) to access the API
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # WARNING: Use specific origins in production (e.g., ["http://localhost:3000", "https://your-frontend-domain.com"])
+    allow_credentials=True,
+    allow_methods=["*"],  # Allow all HTTP methods (GET, POST, PUT, DELETE, etc.)
+    allow_headers=["*"],  # Allow all headers
+)
+
+# --- Database Connection Lifecycle ---
+
+# Connect to the database when the application starts up
+@app.on_event("startup")
+async def startup_event():
+    await connect_db()
+
+# Disconnect from the database when the application shuts down
+@app.on_event("shutdown")
+async def shutdown_event():
+    await disconnect_db()
+
+
+# --- Routes for ESP32 Devices ---
+
+@app.post("/api/devices/register", response_model=DeviceResponse, summary="Register or re-register an ESP32 device")
+async def register_device(device: DeviceRegister):
+    """
+    Registers a new ESP32 device or updates an existing one using 'databases'.
+    Returns a unique API key for the device to use for subsequent requests.
+    """
+    # Call the crud function, which now uses 'databases'
+    return await crud.register_device(device)
+
+@app.post("/api/scan", summary="Receive tag scan data from an ESP32 device and return clearance details")
+async def scan_tag(scan_data: TagScan):
+    """
+    Receives a tag ID from an ESP32 device, verifies the device's API key
+    using 'databases', logs the scan event, retrieves the student's
+    clearance details based on the tag ID, and returns the details to the device.
+    """
+    # Verify API key using the dependency - this will raise HTTPException if invalid
+    # The verify_api_key function now uses 'databases'
+    device = await verify_api_key(scan_data.api_key)
+
+    # Update device last seen timestamp using the crud function (which uses databases)
+    await crud.update_device_last_seen(scan_data.device_id)
+
+    # Log the scan event using the crud function (which uses databases)
+    await crud.create_device_log(scan_data.device_id, scan_data.tag_id, "scan")
+
+    # Get student information by tag ID using the crud function (which uses databases)
+    student = await crud.get_student_by_tag_id(scan_data.tag_id)
+
+    if not student:
+        # Return a specific message to the device if student is not found
+        # The ESP32 firmware should handle this 404 response
+        raise HTTPException(status_code=404, detail="Student not found for this tag")
+
+    # Get clearance statuses for the student using the crud function (which uses databases)
+    statuses = await crud.get_clearance_statuses_by_student_id(student["student_id"])
+
+    # Format clearance items and determine overall status
+    clearance_items = []
+    overall_status = True
+
+    for status_item in statuses:
+        clearance_items.append({
+            "department": status_item["department"],
+            "status": status_item["status"],
+            "remarks": status_item["remarks"],
+            "updated_at": status_item["updated_at"].isoformat() # Format datetime to ISO string
+        })
+        if not status_item["status"]:
+            overall_status = False
+
+    # Return the formatted clearance details
+    return {
+        "student_id": student["student_id"],
+        "name": student["name"],
+        "department": student["department"],
+        "clearance_items": clearance_items,
+        "overall_status": overall_status
+    }
+
+# --- Student Management Routes (Likely for Frontend/Admin Use) ---
+
+@app.post("/api/students/", response_model=Student, status_code=status.HTTP_201_CREATED, summary="Create a new student")
+async def create_student(student: StudentCreate):
+    """Creates a new student record using 'databases'."""
+    # Check if student ID already exists using the crud function (which uses databases)
+    existing_student_by_id = await crud.get_student_by_student_id(student.student_id)
+    if existing_student_by_id:
+        raise HTTPException(status_code=400, detail="Student ID already registered")
+
+    # Check if tag ID already exists using the crud function (which uses databases)
+    existing_student_by_tag = await crud.get_student_by_tag_id(student.tag_id)
+    if existing_student_by_tag:
+        raise HTTPException(status_code=400, detail="Tag ID already assigned to another student")
+
+    # Create the student using the crud function (which uses databases)
+    return await crud.create_student(student)
+
+@app.get("/api/students/", response_model=List[Student], summary="Get all students")
+async def get_students():
+    """Retrieves a list of all students using 'databases'."""
+    # Get all students using the crud function (which uses databases)
+    return await crud.get_all_students()
+
+@app.get("/api/students/{student_id}", response_model=ClearanceDetail, summary="Get clearance details for a specific student")
+async def get_student_clearance(student_id: str):
+    """
+    Retrieves the full clearance details for a student based on their student ID
+    using 'databases'. This endpoint is likely used by the frontend/admin interface.
+    """
+    # Get student info using the crud function (which uses databases)
+    student = await crud.get_student_by_student_id(student_id)
+
+    if not student:
+        raise HTTPException(status_code=404, detail="Student not found")
+
+    # Get clearance status using the crud function (which uses databases)
+    statuses = await crud.get_clearance_statuses_by_student_id(student_id)
+
+    # Format clearance items and determine overall status
+    clearance_items = []
+    overall_status = True
+
+    for status_item in statuses:
+        clearance_items.append({
+            "department": status_item["department"],
+            "status": status_item["status"],
+            "remarks": status_item["remarks"],
+            "updated_at": status_item["updated_at"].isoformat() # Format datetime to ISO string
+        })
+        if not status_item["status"]:
+            overall_status = False
+
+    # Return the formatted clearance details
+    return {
+        "student_id": student["student_id"],
+        "name": student["name"],
+        "department": student["department"],
+        "clearance_items": clearance_items,
+        "overall_status": overall_status
+    }
+
+# --- Clearance Management Routes (Likely for Frontend/Admin Use) ---
+
+@app.post("/api/clearance/", response_model=ClearanceStatus, summary="Create or update a student's clearance status for a department")
+async def update_clearance_status(status_data: ClearanceStatusCreate):
+    """
+    Creates a new clearance status entry for a student and department,
+    or updates an existing one using 'databases'.
+    """
+    # Check if student exists using the crud function (which uses databases)
+    student = await crud.get_student_by_student_id(status_data.student_id)
+    if not student:
+        raise HTTPException(status_code=404, detail="Student not found")
+
+    # Create or update the clearance status using the crud function (which uses databases)
+    return await crud.create_or_update_clearance_status(status_data)
+
+
+# --- Root Endpoint (Optional) ---
+@app.get("/", summary="Root endpoint")
+async def read_root():
+    """Basic root endpoint to confirm the API is running."""
+    return {"message": "Undergraduate Clearance System API is running"}
+
+
+# --- Run the FastAPI app ---
+if __name__ == "__main__":
+    # Use uvicorn to run the FastAPI application
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

requirements.txt

@@ -0,0 +1,11 @@
+fastapi 
+uvicorn 
+databases[postgresql] 
+asyncpg
+pydantic
+pydantic[email]
+pydantic[email,cryptography]
+python-dotenv 
+sqlalchemy
+supabase
+psycopg2-binary

src/auth.py

@@ -0,0 +1,30 @@
+from fastapi import HTTPException, status, Depends
+from src.database import database, devices # Import database instance and devices table
+
+# Dependency to verify the API key sent by ESP32 devices
+async def verify_api_key(api_key: str = Depends(HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing API key"))):
+    """
+    Verifies if the provided API key is valid and corresponds to a registered device
+    using the 'databases' library.
+
+    Args:
+        api_key: The API key provided in the request header or body.
+
+    Returns:
+        The database record of the device if the API key is valid.
+
+    Raises:
+        HTTPException: If the API key is invalid or not found.
+    """
+    # Use 'databases' to execute a select query on the 'devices' table
+    query = devices.select().where(devices.c.api_key == api_key)
+    device = await database.fetch_one(query) # fetch_one returns a single record or None
+
+    if not device:
+        # Raise HTTPException if the API key is not found
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid API key",
+            headers={"WWW-Authenticate": "Bearer"}, # Optional: Suggest Bearer token scheme
+        )
+    return device

src/crud.py

@@ -0,0 +1,151 @@
+# Import database instance and tables from database.py
+from src.database import database, students, clearance_statuses, device_logs, devices
+from src.models import StudentCreate, ClearanceStatusCreate, TagScan, DeviceRegister # Import Pydantic models
+# Import necessary SQLAlchemy components for building queries
+from sqlalchemy import select, insert, update
+from datetime import datetime
+import secrets # For generating API keys
+from fastapi import HTTPException # Import HTTPException to raise errors
+
+# --- CRUD operations for Students ---
+
+async def create_student(student: StudentCreate):
+    """Creates a new student record in the database using 'databases'."""
+    # Build an insert query using SQLAlchemy
+    query = students.insert().values(
+        student_id=student.student_id,
+        name=student.name,
+        department=student.department,
+        tag_id=student.tag_id
+    )
+    # Execute the query asynchronously using 'databases'
+    last_record_id = await database.execute(query)
+    # Return the created student data including the generated ID
+    return {**student.dict(), "id": last_record_id}
+
+async def get_all_students():
+    """Retrieves all student records from the database using 'databases'."""
+    # Build a select query using SQLAlchemy
+    query = students.select()
+    # Execute the query and fetch all results asynchronously
+    return await database.fetch_all(query)
+
+async def get_student_by_student_id(student_id: str):
+    """Retrieves a student record by their student ID using 'databases'."""
+    # Build a select query with a where clause
+    query = students.select().where(students.c.student_id == student_id)
+    # Execute the query and fetch one result asynchronously
+    return await database.fetch_one(query)
+
+async def get_student_by_tag_id(tag_id: str):
+    """Retrieves a student record by their tag ID using 'databases'."""
+    # Build a select query with a where clause
+    query = students.select().where(students.c.tag_id == tag_id)
+    # Execute the query and fetch one result asynchronously
+    return await database.fetch_one(query)
+
+# --- CRUD operations for Clearance Statuses ---
+
+async def create_or_update_clearance_status(status_data: ClearanceStatusCreate):
+    """Creates a new clearance status or updates an existing one using 'databases'."""
+    # Check if clearance status already exists for this student and department
+    query = select(clearance_statuses).where(
+        (clearance_statuses.c.student_id == status_data.student_id) &
+        (clearance_statuses.c.department == status_data.department)
+    )
+    existing_status = await database.fetch_one(query)
+
+    current_time = datetime.utcnow()
+
+    if existing_status:
+        # Update existing status
+        query = update(clearance_statuses).where(
+            (clearance_statuses.c.student_id == status_data.student_id) &
+            (clearance_statuses.c.department == status_data.department)
+        ).values(
+            status=status_data.status,
+            remarks=status_data.remarks,
+            updated_at=current_time
+        )
+        await database.execute(query)
+        # Return the updated record including the existing ID and new timestamp
+        return {
+            **status_data.dict(),
+            "id": existing_status["id"],
+            "updated_at": current_time
+        }
+    else:
+        # Create new status
+        query = insert(clearance_statuses).values(
+            student_id=status_data.student_id,
+            department=status_data.department,
+            status=status_data.status,
+            remarks=status_data.remarks,
+            updated_at=current_time
+        )
+        last_record_id = await database.execute(query)
+        # Return the newly created record including the new ID and timestamp
+        return {
+            **status_data.dict(),
+            "id": last_record_id,
+            "updated_at": current_time
+        }
+
+
+async def get_clearance_statuses_by_student_id(student_id: str):
+    """Retrieves all clearance statuses for a given student ID using 'databases'."""
+    query = select(clearance_statuses).where(clearance_statuses.c.student_id == student_id)
+    return await database.fetch_all(query)
+
+# --- CRUD operations for Devices ---
+
+async def register_device(device_data: DeviceRegister):
+    """Registers a new device or updates an existing one, generating an API key using 'databases'."""
+    api_key = secrets.token_hex(16) # Generate a unique API key
+    current_time = datetime.utcnow()
+
+    # Check if device already exists
+    query = select(devices).where(devices.c.device_id == device_data.device_id)
+    existing_device = await database.fetch_one(query)
+
+    if existing_device:
+        # Update existing device
+        query = update(devices).where(devices.c.device_id == device_data.device_id).values(
+            location=device_data.location,
+            api_key=api_key, # Assign a new API key on re-registration
+            last_seen=current_time
+        )
+        await database.execute(query)
+    else:
+        # Create new device
+        query = insert(devices).values(
+            device_id=device_data.device_id,
+            location=device_data.location,
+            api_key=api_key,
+            last_seen=current_time
+        )
+        await database.execute(query)
+
+    # Return the device details including the generated API key
+    # We need to fetch the newly created/updated device to get the assigned API key
+    query = select(devices).where(devices.c.device_id == device_data.device_id)
+    updated_device = await database.fetch_one(query)
+    return {"device_id": updated_device["device_id"], "location": updated_device["location"], "api_key": updated_device["api_key"]}
+
+
+async def update_device_last_seen(device_id: str):
+    """Updates the 'last_seen' timestamp for a given device using 'databases'."""
+    query = update(devices).where(devices.c.device_id == device_id).values(last_seen=datetime.utcnow())
+    await database.execute(query)
+
+# --- CRUD operations for Device Logs ---
+
+async def create_device_log(device_id: str, tag_id: str, action: str):
+    """Creates a log entry for device activity using 'databases'."""
+    query = device_logs.insert().values(
+        device_id=device_id,
+        tag_id=tag_id,
+        timestamp=datetime.utcnow(),
+        action=action
+    )
+    await database.execute(query)

src/database.py

@@ -0,0 +1,100 @@
+import databases
+import sqlalchemy
+import os
+from dotenv import load_dotenv
+from datetime import datetime
+
+# Load environment variables from a .env file
+load_dotenv()
+
+# Database setup
+# Get the database URL from environment variables
+# IMPORTANT: Ensure your .env file has the correct DATABASE_URL for PostgreSQL
+# Example: DATABASE_URL="postgresql+asyncpg://user:password@host:port/database_name"
+DATABASE_URL = os.getenv("POSTGRES_URI") or os.getenv("DATABASE_URL")
+
+# Check if DATABASE_URL is set
+if not DATABASE_URL:
+    raise ValueError("DATABASE_URL environment variable not set!")
+
+# Create a Database instance for asynchronous operations
+# The 'databases' library uses the connection string to determine the dialect and driver
+database = databases.Database(DATABASE_URL)
+
+# Create a MetaData object to hold the database schema
+metadata = sqlalchemy.MetaData()
+
+# Define the 'students' table using SQLAlchemy
+students = sqlalchemy.Table(
+    "students",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
+    sqlalchemy.Column("student_id", sqlalchemy.String, unique=True, index=True), # Added index for faster lookups
+    sqlalchemy.Column("name", sqlalchemy.String),
+    sqlalchemy.Column("department", sqlalchemy.String),
+    sqlalchemy.Column("tag_id", sqlalchemy.String, unique=True, index=True), # Added index
+)
+
+# Define the 'clearance_statuses' table using SQLAlchemy
+clearance_statuses = sqlalchemy.Table(
+    "clearance_statuses",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
+    sqlalchemy.Column("student_id", sqlalchemy.String, index=True), # Added index
+    sqlalchemy.Column("department", sqlalchemy.String, index=True), # Added index
+    sqlalchemy.Column("status", sqlalchemy.Boolean, default=False),
+    sqlalchemy.Column("remarks", sqlalchemy.String, nullable=True),
+    sqlalchemy.Column("updated_at", sqlalchemy.DateTime, default=datetime.utcnow),
+    # Add a unique constraint to prevent duplicate entries for the same student and department
+    sqlalchemy.UniqueConstraint('student_id', 'department', name='uq_student_department')
+)
+
+# Define the 'device_logs' table to log device activity using SQLAlchemy
+device_logs = sqlalchemy.Table(
+    "device_logs",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
+    sqlalchemy.Column("device_id", sqlalchemy.String, index=True), # Added index
+    sqlalchemy.Column("tag_id", sqlalchemy.String, index=True), # Added index
+    sqlalchemy.Column("timestamp", sqlalchemy.DateTime, default=datetime.utcnow),
+    sqlalchemy.Column("action", sqlalchemy.String), # e.g., "scan", "register"
+)
+
+# Define the 'devices' table to manage registered ESP32 devices using SQLAlchemy (Location Removed)
+devices = sqlalchemy.Table(
+    "devices",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
+    sqlalchemy.Column("device_id", sqlalchemy.String, unique=True, index=True), # Added index
+    # Removed: sqlalchemy.Column("location", sqlalchemy.String),
+    sqlalchemy.Column("api_key", sqlalchemy.String, unique=True, index=True), # Added index for quick API key lookups
+    sqlalchemy.Column("last_seen", sqlalchemy.DateTime, nullable=True),
+)
+
+# Create a SQLAlchemy engine (used for creating tables - typically run once)
+# Note: For PostgreSQL, check_same_thread=False is not needed.
+# This engine is primarily used here for the metadata.create_all call.
+engine = sqlalchemy.create_engine(DATABASE_URL)
+
+try:
+    print("Attempting to create database tables (if they don't exist)...")
+    metadata.create_all(engine)
+    print("Database tables creation attempt finished.")
+except Exception as e:
+    print(f"Error during database table creation: {e}")
+    print("Please ensure your Supabase database is accessible and the connection string is correct.")
+    print("You might need to manually create tables in Supabase using the provided SQL script.")
+
+
+# Database connection lifecycle events (to be called by FastAPI)
+async def connect_db():
+    """Connects to the database on application startup."""
+    print("Connecting to database...") # Added print statement for debugging
+    await database.connect()
+    print("Database connected.") # Added print statement for debugging
+
+async def disconnect_db():
+    """Disconnects from the database on application shutdown."""
+    print("Disconnecting from database...") # Added print statement for debugging
+    await database.disconnect()
+    print("Database disconnected.") # Added print statement for debugging

src/models.py

@@ -0,0 +1,52 @@
+from pydantic import BaseModel
+from typing import List, Optional
+from datetime import datetime
+
+# Pydantic models for data validation and serialization
+
+class StudentCreate(BaseModel):
+    """Model for creating a new student."""
+    student_id: str
+    name: str
+    department: str
+    tag_id: str
+
+class Student(StudentCreate):
+    """Model for returning student data, including database ID."""
+    id: int
+
+class ClearanceStatusCreate(BaseModel):
+    """Model for creating or updating a student's clearance status for a specific department."""
+    student_id: str
+    department: str
+    status: bool
+    remarks: Optional[str] = None
+
+class ClearanceStatus(ClearanceStatusCreate):
+    """Model for returning clearance status data, including database ID and update timestamp."""
+    id: int
+    updated_at: datetime
+
+class ClearanceDetail(BaseModel):
+    """Model for returning a student's full clearance details."""
+    student_id: str
+    name: str
+    department: str
+    clearance_items: List[dict] # List of dictionaries representing each clearance item
+    overall_status: bool
+
+class DeviceRegister(BaseModel):
+    """Model for registering a new ESP32 device."""
+    device_id: str
+    location: str
+
+class DeviceResponse(DeviceRegister):
+    """Model for returning device registration details, including the API key."""
+    api_key: str
+
+class TagScan(BaseModel):
+    """Model for data received from an ESP32 device after scanning a tag."""
+    device_id: str
+    tag_id: str
+    api_key: str
+    timestamp: datetime