prepAI / Readme.md
aki-008's picture
feat: testing new endpoint
6b609bb
|
raw
history blame
45.5 kB

🔍 FastAPI + PostgreSQL Student Management System - Line-by-Line Explainer

Complete breakdown of an async student API with authentication, explained like you're learning it for the first time.


📚 Table of Contents

  1. Overview & Imports
  2. Configuration
  3. Database Setup
  4. Database Models
  5. Pydantic Validation Models
  6. Security & Authentication
  7. Database Dependency
  8. FastAPI Application
  9. Startup & Shutdown Events
  10. Authentication Routes
  11. Student CRUD Operations
  12. Health Check & Runner

1️⃣ Overview & Imports

Comment Block

"""
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
"""

What it is:
A multi-line docstring (document string) that acts as the script's billboard and shopping list.

What it's used for:
This is like the "Welcome to our restaurant" sign that tells you:

  1. What you're building - A student management API
  2. The tech stack - FastAPI, PostgreSQL, async everything
  3. The installation cheat-sheet - Copy-paste commands to get all required tools

IRL Example:
Imagine buying furniture from IKEA. This comment block is the box label that says "BILLY Bookcase" and lists all the screws, tools, and instructions inside. Without it, you'd have no idea what you're about to build or what tools you need.


FastAPI Imports

from fastapi import FastAPI, HTTPException, Depends, status

What it is:
Importing the four core ingredients to build a FastAPI application.

What it's used for:

  • FastAPI: The main framework (your restaurant's foundation and kitchen layout)
  • HTTPException: Standardized error messages (your "Sorry, we're out of salmon" response)
  • Depends: Dependency injection system (your "I need a clean plate from the dishwasher" request)
  • status: HTTP status code constants (your "404 = Not Found" cheat sheet)

Syntax Breakdown:
Python's from ... import ... grabs specific tools from a toolbox instead of lugging the entire toolbox around.

IRL Example:
Opening a restaurant. Instead of renting an entire warehouse of kitchen equipment (import fastapi), you specifically rent: a commercial oven (FastAPI), a fire alarm (HTTPException), a dishwashing service (Depends), and a health code manual (status).


Security Imports

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

What it is:
Tools for implementing token-based authentication.

What it's used for:

  • HTTPBearer: The security scheme that looks for "Authorization: Bearer " headers
  • HTTPAuthorizationCredentials: The container that holds the extracted token

IRL Example:
A nightclub's security system. HTTPBearer is the policy that says "We only accept VIP badges with a hologram strip," and HTTPAuthorizationCredentials is the scanner device that reads and validates those badges at the door.


CORS Middleware

from fastapi.middleware.cors import CORSMiddleware

What it is:
Cross-Origin Resource Sharing (CORS) protection.

What it's used for:
Prevents malicious websites from making unauthorized requests to your API from a browser. It's like a "no solicitation" sign that tells browsers which domains are allowed to knock on your API's door.

IRL Example:
You own a house (your API). CORSMiddleware is your doorbell camera that checks if the visitor is from your approved guest list (allowed_origins). Without it, any stranger could walk up and demand entry.


SQLAlchemy Async Imports

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

What it is:
The async-capable database tools from SQLAlchemy.

What it's used for:

  • create_async_engine: Builds a non-blocking database connection pool
  • AsyncSession: A database session that won't freeze your API while waiting for queries
  • async_sessionmaker: Factory that produces these async sessions on demand

IRL Example:
A modern coffee shop with mobile ordering. Instead of one cashier handling one customer at a time (sync), you have a system that:

  • Takes orders (AsyncSession)
  • Processes multiple orders simultaneously without waiting for each drink to finish (async)
  • Has multiple baristas ready to make drinks (connection pool)

SQLAlchemy ORM Imports

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

What it is:
Object-Relational Mapping (ORM) tools that let you use Python classes as database tables.

What it's used for:

  • DeclarativeBase: The foundation class for all your database models
  • Mapped: Type hint that says "this class attribute maps to a database column"
  • mapped_column: Factory function to configure column properties

IRL Example:
A digital Rolodex system. Instead of writing raw SQL like INSERT INTO students VALUES (...), you create a Student class (DeclarativeBase) where each property (Mapped[name]) automatically syncs with a database column, like magic index cards that file themselves.


SQLAlchemy Query & Types

from sqlalchemy import select, String

What it is:
Essential SQL building blocks.

What it's used for:

  • select: Creates SELECT queries in Python (instead of raw SQL strings)
  • String: Defines VARCHAR columns in PostgreSQL

IRL Example:
select is like a smart query assistant who speaks both Python and SQL. Instead of handwriting "SELECT * FROM students WHERE age > 18", you tell it in Python: select(Student).where(Student.age > 18) and it translates perfectly.


Pydantic Imports

from pydantic import BaseModel, EmailStr, Field, validator

What it is:
Data validation and serialization library that acts as your API's security guard and translator.

What it's used for:

  • BaseModel: Foundation for all data models
  • EmailStr: Automatic email format validation
  • Field: Fine-tuned validation rules (min/max length, descriptions)
  • validator: Custom validation functions (your special rules)

IRL Example:
A TSA checkpoint at an airport. BaseModel is the checkpoint itself. EmailStr is the automated scanner that knows a valid passport format. Field is the sign saying "Liquids must be < 100ml." validator is the agent who says "Sir, you can't bring a full water bottle through."


Typing & Utility Imports

from typing import Optional, List
from datetime import datetime, timedelta

What it is:
Type hints and date/time utilities.

What it's used for:

  • Optional[T]: Means "this value is either type T or None" (Optional[str] = string or None)
  • List[T]: Type hint for lists of a specific type
  • datetime: Current timestamp
  • timedelta: Time arithmetic (30 minutes from now)

IRL Example:
A delivery tracking app. Optional[str] handles "delivery instructions" that might be empty. List[Package] ensures you're dealing with packages, not random objects. datetime stamps when it shipped, and timedelta calculates "expected delivery in 3 days."


JWT & Password Imports

from jose import JWTError, jwt
from passlib.context import CryptContext

What it is:
JSON Web Token handling and password hashing libraries.

What it's used for:

  • jwt.encode/decode: Creates and verifies authentication tokens
  • JWTError: Exception when tokens are invalid/expired
  • CryptContext: Manages password hashing algorithms

IRL Example:
A concert venue's wristband system. jwt.encode prints a tamper-proof wristband with your access level. jwt.decode is the UV scanner at each entrance. CryptContext is the secure machine that embosses wristbands so they can't be faked.


OS Import

import os

What it is:
Python's interface to operating system environment variables.

What it's used for:

  • os.getenv(): Safely reads environment variables (database passwords, secret keys)
  • Keeps sensitive data out of your source code

IRL Example:
A hotel key card system. The master key code (SECRET_KEY) is stored in the hotel safe (environment variables), not written on a sticky note on the front desk (hardcoded in script). os.getenv() is the manager opening the safe when needed.


2️⃣ Configuration

Database URL Configuration

DATABASE_URL = os.getenv(
    "DATABASE_URL", 
    "postgresql+asyncpg://postgres:690869@172.26.157.164:5432/studentdb"
)

What it is:
Environment-driven database connection string with a fallback default.

What it's used for:

  • os.getenv() first tries to read from system environment variables (production best practice)
  • Fallback string is used if environment variable doesn't exist (development convenience)
  • Format: dialect+driver://user:password@host:port/database

Syntax Breakdown:

  • postgresql+asyncpg://: Use PostgreSQL with asyncpg driver for non-blocking operations
  • postgres:690869: Username and password (⚠️ Never commit real passwords!)
  • @172.26.157.164:5432: Database server IP and port
  • /studentdb: Database name

IRL Example:
A smart home door lock. You program it to first check if there's a master code set via the mobile app (os.getenv()). If not, it falls back to the factory default code (fallback string). In production, you always use the mobile app; the default is just for initial setup.


JWT Secret Configuration

SECRET_KEY = os.getenv("SECRET_KEY", "production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

What it is:
Security parameters for JWT token generation.

What it's used for:

  • SECRET_KEY: The master password used to sign tokens (keeps them tamper-proof)
  • ALGORITHM: HS256 is a symmetric signing algorithm (same key signs and verifies)
  • ACCESS_TOKEN_EXPIRE_MINUTES: Tokens auto-expire after 30 minutes for security

IRL Example:
A theme park's ride pass system:

  • SECRET_KEY is the unique holographic foil pattern only your park possesses
  • ALGORITHM is the type of embossing machine (HS256 = standard heat press)
  • ACCESS_TOKEN_EXPIRE_MINUTES is the stamp that says "Valid only today until 6 PM"

3️⃣ Database Setup

Async Engine Creation

engine = create_async_engine(DATABASE_URL, echo=True)

What it is:
The heart of your async database connection pool.

What it's used for:

  • Creates a non-blocking connection engine that handles multiple requests simultaneously
  • echo=True logs all SQL queries to console (invaluable for debugging)

IRL Example:
A high-end restaurant's kitchen. engine is the head chef who:

  • Manages multiple cooking stations (connections)
  • Can start prep for Order #2 while Order #1 is in the oven (async)
  • Wears a bodycam (echo=True) so management can review exactly what happened during the dinner rush

Session Factory

async_session_maker = async_sessionmaker(engine, expire_on_commit=False)

What it is:
A factory that produces database sessions on demand.

What it's used for:

  • expire_on_commit=False: Keeps data in memory after committing (prevents unnecessary re-queries)
  • Each API request gets its own session (like giving each customer a fresh plate)

IRL Example:
A conveyor belt sushi restaurant. The async_session_maker is the automated plate dispenser. Every time a customer sits down (API request), it dispenses a fresh, clean plate (session). The plate stays usable even after they take their first sushi (commit), so they don't have to get a new plate for every single bite.


4️⃣ Database Models

Base Model Foundation

class Base(DeclarativeBase):
    pass

What it is:
The foundation class that all database models inherit from.

What it's used for:

  • SQLAlchemy needs a central Base to register all tables
  • Even though it's empty (pass), it's crucial for metadata collection

IRL Example:
A filing cabinet's index system. Every type of document (Student, User) has its own folder, but they all share the same index card system (Base). The index system doesn't contain data itself but knows where everything is stored.


Student Model

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)

What it is:
A Python class that magically becomes a PostgreSQL table.

What it's used for:

  • __tablename__: The actual table name in PostgreSQL
  • Each Mapped[column] becomes a column with constraints

Syntax Breakdown:

  • id: Primary key, auto-incrementing, indexed for fast lookups
  • name: String up to 100 characters
  • email: String, must be unique across all students, indexed for fast email searches
  • age: Plain integer
  • grade: String up to 5 chars (stores "A+", "F", etc.)
  • created_at: Auto-set to current UTC time when record is created

IRL Example:
A school registrar's digital filing system. Instead of filling out paper forms, they use a tablet app where:

  • Each field is validated before submission
  • Student ID is auto-generated and used for quick filing
  • Email must be unique (can't have two students with same email)
  • Timestamp is automatically stamped when the form is first saved

User Model

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))

What it is:
A separate table for storing authenticated user accounts.

What it's used for:

  • Stores login credentials separately from student data
  • hashed_password contains scrambled passwords (never plain text!)
  • 255 char limit accommodates long Argon2 hashes

IRL Example:
A school's staff keycard system. There's a locked cabinet (users table) containing:

  • Employee ID numbers (id)
  • Name badges (username, unique)
  • Encrypted keycard codes (hashed_password) that can't be reverse-engineered

This is separate from the student records office—teachers have access, but the data is stored in different secure locations.


5️⃣ Pydantic Validation Models

StudentBase Model

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 -)")

What it is:
The foundational validation blueprint for student data.

What it's used for:

  • Field(...): Makes the field required (can't be omitted)
  • min_length=2: Name must be at least 2 characters
  • ge=5, le=100: Age must be between 5 and 100 (inclusive)
  • pattern: Regular expression ensuring grade matches letter grades only

IRL Example:
A strict school application kiosk:

  • Name field: Won't accept "J" or just spaces; minimum 2 real letters
  • Email field: Has an @ symbol, valid domain; rejects "bob@com"
  • Age field: Slider only goes from 5 to 100; toddlers can't apply
  • Grade field: Dropdown only shows valid grades (A+, A, A-, B+, etc.)

Name Validator

@validator('name')
def validate_name(cls, v):
    if not v.strip():
        raise ValueError('Name cannot be empty or just whitespace')
    return v.strip()

What it is:
Custom validation logic for the name field.

What it's used for:

  • v.strip(): Removes leading/trailing spaces
  • Raises error if the result is empty (e.g., name was just spaces)
  • Returns the cleaned version

IRL Example:
A DMV form validator. When you type " John " in the name field, it automatically trims the spaces to "John" before submitting. If you try to submit just spaces, the system beeps and says "Please enter a valid name."


Grade Validator

@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

What it is:
Ensures only valid letter grades are accepted.

What it's used for:

  • Converts input to uppercase (a+A+)
  • Checks against an exhaustive whitelist
  • Prevents typos like "A++" or "B--"

IRL Example:
A Scantron machine. When teachers grade multiple-choice tests, the machine only accepts certain bubbled answers (A, B, C, D, E). If a student bubbles "F" on a 5-question test, the machine rejects it as invalid.


StudentCreate Model

class StudentCreate(StudentBase):
    pass

What it is:
A clone of StudentBase for creating new students.

What it's used for:

  • Inheritance: Gets all validation from parent
  • Semantic clarity: StudentCreate tells developers this is for creation only
  • Future-proofing: You can add creation-specific fields later without breaking existing code

IRL Example:
A "New Student Registration Form" that's identical to the regular form but printed on blue paper. Today it's the same, but tomorrow you might add a "Parent Signature" field that's only on the blue creation form.


StudentUpdate Model

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][+-]?$")

What it is:
A model for partial student updates where all fields are optional.

What it's used for:

  • Optional[T] = None: Every field can be omitted (PATCH request style)
  • Allows updating just one field (e.g., only change grade) without sending full data

IRL Example:
A student profile editing page on a school portal. You can change just the grade without re-entering name, email, and age. The form has "Save" buttons next to individual fields—all fields are optional, but if you fill one, it must still pass validation.


StudentResponse Model

class StudentResponse(StudentBase):
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True

What it is:
The response format when sending student data back to clients.

What it's used for:

  • Adds database-generated fields (id, created_at) to the base model
  • from_attributes = True: Tells Pydantic to read from SQLAlchemy object attributes (.id, .name) instead of dictionary keys (["id"], ["name"])

IRL Example:
A school report card printing system. The blank form (StudentBase) has name, grade, etc. The printed report (StudentResponse) adds the official student ID number and the date issued—fields the student didn't provide but were generated by the system.


UserCreate Model

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    password: str = Field(..., min_length=6, max_length=72)

What it is:
Validation model for new user registration.

What it's used for:

  • min_length=6: Basic password strength requirement
  • max_length=72: Critical for bcrypt/Argon2 compatibility (they truncate at 72 bytes)

Password Byte-Size Validator

@validator('password')
def validate_password(cls, v):
    if len(v.encode('utf-8')) > 72:
        raise ValueError('Password cannot exceed 72 bytes')
    return v

What it is:
Ensures password doesn't exceed 72 bytes (not characters—important for Unicode!).

What it's used for:

  • Some characters (like emojis) use multiple bytes
  • "🔒🔒🔒" might be 3 characters but 12 bytes
  • Prevents silent truncation by bcrypt

IRL Example:
A text message limit. You can send 160 characters, but if you use special Unicode symbols (like 🎉), each might count as 2-4 characters. The validator warns you before you exceed the hidden byte limit.


Token Response Model

class Token(BaseModel):
    access_token: str
    token_type: str

What it is:
The login response structure.

What it's used for:

  • Standard OAuth2 token format
  • access_token: The JWT string
  • token_type: Always "bearer" for Bearer token authentication

IRL Example:
A valet ticket. It has a unique code (access_token) and the type of service (token_type: "bearer") which tells the API "This is a bearer token, not a basic auth or API key."


6️⃣ Security & Authentication

Password Context

pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")

What it is:
Configures Argon2 as the password hashing algorithm.

What it's used for:

  • Argon2 is the winner of the Password Hashing Competition (resistant to GPU attacks)
  • deprecated="auto": Future-proofs against algorithm upgrades

IRL Example:
Choosing a safe for your vault. You select the Argon2 model because it's drill-resistant, fireproof, and can't be opened with a stethoscope (GPU cracking). The "auto-deprecate" feature means if a better safe comes out, the system automatically flags old ones as outdated.


Password Verification Function

def verify_password(plain_password: str, hashed_password: str) -> bool:
    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)

What it is:
Safely checks if a plain password matches a stored hash.

What it's used for:

  • Truncates to 72 bytes before verifying (prevents timing attacks)
  • errors='ignore': Drops invalid byte sequences gracefully
  • Returns True if match, False otherwise

IRL Example:
A high-security office keypad. When you type your code:

  1. The system only reads first 10 digits (ignores extra)
  2. Compares against stored scrambled code (not the real code)
  3. Green light if they match, red light if not
  4. Takes same time whether you typed 6 or 10 digits (prevents guessing)

Password Hashing Function

def get_password_hash(password: str) -> str:
    password_bytes = password.encode('utf-8')[:72]
    password_truncated = password_bytes.decode('utf-8', errors='ignore')
    return pwd_context.hash(password_truncated)

What it is:
Converts plain passwords into irreversible hashes for storage.

What it's used for:

  • Never store plain passwords
  • Same truncation logic as verification
  • Returns a long string starting with $argon2...

IRL Example:
A paper shredder that creates confetti patterns. Every time you shred "password123", it creates the same unique confetti pattern. But you can't reverse the pattern back to "password123". Even if thieves steal the confetti, they can't reconstruct the original document.


JWT Token Creation

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

What it is:
Generates a time-limited authentication token.

What it's used for:

  • Copies user data (like username)
  • Adds expiration timestamp (exp claim)
  • Signs with secret key to prevent tampering
  • Returns base64-encoded JWT string

IRL Example:
A concert venue wristband machine:

  1. You input data: "Adult, 21+, VIP" (data: dict)
  2. Machine stamps expiration: "Valid until 11 PM" (expires_delta)
  3. Embosses with holographic seal (SECRET_KEY + ALGORITHM)
  4. Spits out wristband with barcode (JWT string)

If someone tries to alter "VIP" to "ALL ACCESS", the seal breaks and scanners reject it.


Current User Dependency (The Auth Bouncer)

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

What it is:
The authentication gatekeeper for protected routes.

What it's used for:

  1. Extract token: Gets Authorization: Bearer <token> header
  2. Decode JWT: Verifies signature and expiration
  3. Extract username: From "sub" (subject) claim
  4. Database lookup: Ensures user still exists
  5. Return user object: Injects into route functions

IRL Example:
A corporate building with RFID badge access:

  1. Badge swipe: Employee taps card (credentials)
  2. Scanner verification: Checks if badge is forged/expired (jwt.decode)
  3. Database check: Looks up employee ID to ensure they're still employed (select(User))
  4. Door opens: Returns employee record to security system (return user)
  5. Invalid badge: Alarm sounds, door stays locked (HTTPException)

If badge is expired (JWTError) or employee was fired (user is None), the bouncer yells "Access denied!"


7️⃣ Database Dependency

Session Lifetime Manager

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()

What it is:
A context manager that handles database sessions for each request.

What it's used for:

  • Creates session: async with opens a new session
  • Yields: Gives session to route function (yield session)
  • Auto-commit: If no errors, saves changes (await session.commit())
  • Auto-rollback: If error occurs, undoes changes (await session.rollback())
  • Cleanup: Always closes session (finally: await session.close())

IRL Example:
A bank teller's daily procedure:

  1. Open drawer: Gets cash drawer for the day (yield session)
  2. Process transactions: Helps customers (route function runs)
  3. Balance: If no errors, locks drawer and finalizes (commit)
  4. Error: If robbed, triggers silent alarm and voids transactions (rollback)
  5. Close: Goes home, drawer locked regardless (finally: close)

Even if the teller faints mid-transaction, the finally block ensures the drawer gets locked.


8️⃣ FastAPI Application

App Initialization

app = FastAPI(
    title="Student Management API",
    description="FastAPI + PostgreSQL with SQLAlchemy async",
    version="1.0.0"
)

What it is:
Creates the main FastAPI application instance.

What it's used for:

  • title: Shows in API documentation (Swagger UI)
  • description: Detailed info about the API
  • version: Semantic versioning for API changes

IRL Example:
Opening a new restaurant. You register it with:

  • Name: "Student Management Diner" (title)
  • Description: "Serves fresh student data with PostgreSQL sauce" (description)
  • Version: "Menu v1.0" (version)

This info appears on your website and menu (API docs).


CORS Middleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, specify allowed origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

What it is:
Configures Cross-Origin Resource Sharing policy.

What it's used for:

  • allow_origins=["*"]: DANGEROUS - allows any website to call your API
  • allow_credentials=True: Allows cookies/auth headers
  • allow_methods=["*"]: Allows GET, POST, PUT, DELETE, etc.
  • allow_headers=["*"]: Allows any custom headers

⚠️ Production Warning:
["*"] is like leaving your front door unlocked. In production, specify exact domains like ["https://your-frontend.com"].

IRL Example:
A public library's computer policy vs. a corporate VPN:

  • Library (allow_origins=["*"]): Anyone can walk in and use computers
  • Corporate (allow_origins=["https://corp.com"]): Only company-issued laptops can connect

9️⃣ Startup & Shutdown Events

Startup Event

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("✅ Database tables created successfully")

What it is:
Auto-creates database tables when the app starts.

What it's used for:

  • engine.begin(): Starts a database transaction
  • run_sync: Converts async connection to sync for SQLAlchemy's create_all
  • Creates students and users tables if they don't exist
  • Prints success message

IRL Example:
A pop-up restaurant's opening checklist:

  1. Unlock doors (engine.begin())
  2. Set up tables and chairs (create_all())
  3. Check inventory (tables created)
  4. Unlock sign turns green (print success)

If tables already exist, it's like "Tables are already set up, ready to serve!"


Shutdown Event

@app.on_event("shutdown")
async def shutdown():
    await engine.dispose()
    print("🔴 Database connection closed")

What it is:
Gracefully closes database connections when app stops.

What it's used for:

  • engine.dispose(): Closes all connections in the pool
  • Prevents "orphaned" connections that eat database resources
  • Prints closure message for ops visibility

IRL Example:
A store's closing procedure:

  1. Last customer leaves (last request processed)
  2. Lock doors (shutdown event triggered)
  3. Turn off all lights and equipment (engine.dispose())
  4. Security system armed (connections closed)
  5. Sign on door: "Closed" (print message)

Without this, it's like leaving all lights and AC on overnight—wastes resources.


10️⃣ Authentication Routes

Register Endpoint

@app.post("/auth/register", response_model=dict, tags=["Authentication"])
async def register(user: UserCreate, db: AsyncSession = Depends(get_db)):
    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}

What it is:
Creates a new user account with hashed password.

What it's used for:

  1. Check existence: Queries if username already taken
  2. Hash password: Never stores plain text
  3. Create user: Adds to database
  4. Commit: Saves transaction
  5. Return success: JSON response

IRL Example:
A gym membership signup:

  1. Desk clerk checks if your desired username "GymGuru" is taken
  2. If taken: "Sorry, that username is already in use" (HTTPException)
  3. If available: Scans your ID, takes photo (User object created)
  4. Makes membership card (hashed_password = encrypted member ID)
  5. Welcome package: "Welcome GymGuru! Membership #12345 activated"

Login Endpoint

@app.post("/auth/login", response_model=Token, tags=["Authentication"])
async def login(username: str, password: str, db: AsyncSession = Depends(get_db)):
    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="...")

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"}

What it is:
Verifies credentials and returns JWT token.

What it's used for:

  1. Fetch user: Get user record from database
  2. Verify password: Check hash against input
  3. Create token: Generate JWT with username in "sub" claim
  4. Return: Token for client to store (localStorage, cookies)

IRL Example:
Hotel check-in:

  1. Guest says: "I'm John Doe, room 123" (username, password)
  2. Clerk checks: Looks up reservation (select(User))
  3. ID verification: Scans driver's license (verify_password)
  4. If wrong: "Sorry, we have no reservation under that name" (401)
  5. If correct: Issues room keycard (JWT token) valid for 3 days (timedelta)
  6. Keycard says: "Bearer" (you must bear this card to access your room)

The token is your keycard—show it at the bar, gym, pool (protected endpoints) to prove you're a guest.


11️⃣ Student CRUD Operations

Create Student (Protected)

@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)
):
    result = await db.execute(select(Student).filter(Student.email == student.email))
    if existing_student:
        raise HTTPException(status_code=400, 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

What it is:
Creates a new student record (requires authentication).

What it's used for:

  • Authentication: Depends(get_current_user) ensures user is logged in
  • Email uniqueness: Prevents duplicate emails
  • Dict unpacking: **student.model_dump() converts Pydantic model to SQLAlchemy object
  • Refresh: Reloads from DB to get id and created_at

IRL Example:
A school registrar's terminal (requires staff login):

  1. Staff badge swipe (get_current_user verifies)
  2. Fill form: New student data entered
  3. System checks: "Email already exists? Show error"
  4. If unique: Student record created, ID assigned (auto-increment)
  5. Receipt printed: Shows full student data with ID and enrollment date

Without badge swipe, the terminal shows "Access Denied."


Get All Students (Paginated)

@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)
):
    result = await db.execute(select(Student).offset(skip).limit(limit))
    students = result.scalars().all()
    return students

What it is:
Returns a list of students with pagination.

What it's used for:

  • skip: Number of records to skip (for page 2, 3, etc.)
  • limit: Max records per page (prevents overwhelming the client)
  • result.scalars().all(): Converts result rows into list of Student objects

IRL Example:
A school directory on a website:

  • Page 1: skip=0, limit=100 (students 1-100)
  • Page 2: skip=100, limit=100 (students 101-200)
  • Staff only: Must be logged in to view (get_current_user)
  • Fast: Indexes on id and email make queries speedy

Without pagination, requesting 10,000 students at once could crash the browser.


Get Single Student

@app.get("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
async def get_student(student_id: int, ...):
    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

What it is:
Fetches one student by ID.

What it's used for:

  • student_id: int: Path parameter from URL (/students/42)
  • scalar_one_or_none(): Returns one result or None (not a list)
  • 404 if not found: Standard "not found" response

IRL Example:
A librarian looking up a book:

  1. Patron asks: "Do you have book #12345?" (student_id)
  2. Librarian checks catalog (select(Student))
  3. Book not found: "Sorry, that book isn't in our system" (404)
  4. Book found: Hands over the book details

The ID is like a library card number—unique and used for direct lookup.


Update Student (PUT)

@app.put("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
async def update_student(
    student_id: int,
    student_update: StudentUpdate,
    ...
):
    result = await db.execute(select(Student).filter(Student.id == student_id))
    student = result.scalar_one_or_none()
    if not student:
        raise HTTPException(status_code=404, detail=f"Student with ID {student_id} not found")
    
    update_data = student_update.model_dump(exclude_unset=True)
    if "email" in update_data:
        # Check email uniqueness across other students
        ...
    
    for key, value in update_data.items():
        setattr(student, key, value)
    
    await db.commit()
    await db.refresh(student)
    return student

What it is:
Full or partial update of a student's information.

What it's used for:

  • exclude_unset=True: Only updates fields that were actually sent (PATCH behavior)
  • Email check: Ensures new email doesn't belong to another student
  • setattr(student, key, value): Dynamically updates attributes

IRL Example:
A student filing a change-of-address form:

  1. Submit form: Only fill "New Address" field (other fields omitted)
  2. System checks: "Is this address used by another student?"
  3. Update record: Changes only the address field (setattr)
  4. Save: File updated, timestamp unchanged
  5. Confirmation: Shows full updated record

If you tried to change email to one already in use: "Sorry, that email is taken."


Patch Student Alias

@app.patch("/students/{student_id}", response_model=StudentResponse, tags=["Students"])
async def partial_update_student(...):
    return await update_student(student_id, student_update, db, current_user)

What it is:
A PATCH route that reuses PUT logic.

What it's used for:

  • HTTP semantics: PUT = full replace, PATCH = partial update
  • Here both behave identically due to exclude_unset=True
  • Could be refactored later to have different validation

IRL Example:
A restaurant with two doors:

  • Main entrance (PUT): You can enter with full party or alone
  • Side door (PATCH): Says "Express Entry" but leads to same host stand
  • Both get you to the same table; side door is just for "quick update" perception

Delete Student

@app.delete("/students/{student_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Students"])
async def delete_student(...):
    result = await db.execute(select(Student).filter(Student.id == student_id))
    student = result.scalar_one_or_none()
    if not student:
        raise HTTPException(status_code=404, detail=f"Student with ID {student_id} not found")
    
    await db.delete(student)
    await db.commit()
    return None

What it is:
Removes a student from the database.

What it's used for:

  • 204 No Content: Successful deletion returns no body (REST best practice)
  • Idempotent: Deleting twice is safe (second time returns 404)
  • Return None: FastAPI recognizes None + 204 status = empty response

IRL Example:
A school registrar shredding a withdrawn student's file:

  1. Confirm ID: "Withdraw student #42?" (student_id)
  2. If not found: "No such student record" (404)
  3. If found: Shreds file (db.delete), updates master index (commit)
  4. No receipt: Just a nod ("204")—nothing to show, job done

HTTP 204 is like a paperless receipt: "Success, but nothing to return."


12️⃣ Health Check & Runner

Health Endpoint

@app.get("/", tags=["Health"])
async def root():
    return {
        "status": "healthy",
        "message": "Student Management API is running",
        "version": "1.0.0"
    }

What it is:
A simple endpoint to check if API is alive.

What it's used for:

  • Load balancers ping this to see if instance is healthy
  • Monitoring tools check uptime
  • Humans can curl http://localhost:8000/ for quick test

IRL Example:
A doctor's stethoscope check. The doctor (load balancer) listens to your heart (GET /) and hears:

  • "Status: Healthy!"
  • "Message: All systems operational"
  • "Version: Human v2.0"

If no heartbeat, you're marked "unhealthy" and removed from service.


Application Runner

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

What it is:
Direct execution script for development.

What it's used for:

  • if __name__ == "__main__": Only runs when script is executed directly (not imported)
  • uvicorn.run(): Starts the ASGI server
  • "main:app": String format for reload mode (module:app instance)
  • host="0.0.0.0": Listens on all network interfaces (accessible from other computers)
  • port=8000: HTTP port
  • reload=True: Dev only - auto-restarts on code changes

IRL Example:
Starting a car:

  • if __name__ == "__main__" is the key ignition check (only start if key is turned)
  • uvicorn.run() is the engine starter motor
  • reload=True is like a mechanic watching your engine and immediately restarting it if you tweak the carburetor (code changes)
  • In production, you remove the mechanic (reload=False) for performance

🎯 Key Concepts Summary

Concept IRL Analogy Why It Matters
Async Coffee shop mobile orders Handle many requests without waiting
SQLAlchemy ORM Digital Rolodex Use Python objects, not raw SQL strings
Pydantic Airport TSA Validate everything before it enters
JWT Concert wristband Stateless auth, scalable across servers
CORS Doorbell camera Block unauthorized websites
Dependencies Dishwashing service Reusable, clean resource management
Migrations IKEA assembly manual Version control for database schema

🔒 Security Checklist

  • Never hardcode secrets - Use environment variables
  • Never store plain passwords - Hash with Argon2
  • Never trust client input - Pydantic validates everything
  • Protect all routes - Depends(get_current_user) on sensitive endpoints
  • Use HTTPS in production - Prevents token interception
  • ⚠️ Fix CORS - Change ["*"] to specific domains in production
  • ⚠️ Rate limiting - Add to prevent brute-force on /auth/login
  • ⚠️ Password reset - Add flow for forgotten passwords

🚀 Running the Application

Development:

pip install -r requirements.txt
export DATABASE_URL="postgresql+asyncpg://..."
export SECRET_KEY="your-secret-key-here"
python main.py

Production (using gunicorn):

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Access API Docs: Open browser to http://localhost:8000/docs for interactive Swagger UI.


End of explainer. Happy coding!