Kumar Vipin Yadav

Basics Of FastAPI In Python (SQLModel Edition)

#fastapi #backend #async #sqlmodel

The Comprehensive Guide to Modern API Development with FastAPI and SQLModel

Hello Everyone! Today we are shifting into modern backend design with FastAPI.

Instead of writing classic, verbose object-relational mappings, we are using SQLModel to handle our data persistence. SQLModel is a cutting-edge library that unifies Pydantic and SQLAlchemy into a single data design tool. This means a single Python class acts as both your network validation gate and your physical database layout, eliminating redundant code entirely.

1. The ASGI Server Architecture: Why uvicorn main:app --reload?

FastAPI uses ASGI (Asynchronous Server Gateway Interface), which is designed to handle thousands of concurrent incoming network connections simultaneously.

When you run uvicorn main:app --reload in your terminal, you are launching Uvicorn, a lightning-fast ASGI server wrapper.

Why Do We Run Uvicorn?

  • Asynchronous Concurrency: Traditional synchronous servers block a execution thread whenever a request is waiting on something slow (like an external API call). An ASGI server pauses the waiting request, handles hundreds of other incoming requests in the meantime, and resumes the original request the millisecond the background system answers.
  • The Application Reference (main:app): main tells Uvicorn to look inside your main.py script file, and app points directly to the instantiated FastAPI object inside it.
  • --reload (Hot Swapping): Monitors your code changes. The exact instant you hit save, the Uvicorn workers reboot safely in milliseconds.

2. HTTP Operations, Status Codes, and Routes in FastAPI

FastAPI handles routing explicitly, using standard Python type hints to ensure data validation before it ever enters your backend functions.

Operations and Decorator Syntax

FastAPI maps HTTP actions directly onto decorator methods named after the action itself (e.g., @app.get(), @app.post(), @app.put(), @app.delete()).

Returning Explicit Status Codes

Instead of passing numbers or manual tuples back, FastAPI provides a clean status module that makes your code highly readable.

from fastapi import FastAPI, status

app = FastAPI()

@app.get("/health", status_code=status.HTTP_200_OK)
def health_check():
    return {"status": "operational", "message": "Server is running smoothly."}

Or the same can be done by response module

from fastapi import FastAPI, status, Response

app = FastAPI()

@app.get("/health")
def health_check(response: Response):
    response.status_code = status.HTTP_401_UNAUTHORIZED
    return {"status": "operational", "message": "Server is running smoothly."}

3. Advanced Dynamic Routing and Data Extraction

FastAPI treats path parameters and query parameters with strict structural precision. It reads your function signature and automatically converts and validates incoming data types.

from fastapi import FastAPI

app = FastAPI()

# 1. Dynamic Path Parameter with Type Validation
@app.get("/items/{item_id}")
def read_item(item_id: int):
    # FastAPI automatically throws a clean '422 Unprocessable Entity' error 
    # if a user tries to send a string like "/items/abc" instead of an integer.
    return {"item_id": item_id, "type": str(type(item_id))}

# 2. Query Parameters (Extracted automatically from URL query strings)
@app.get("/search")
def search_database(q: str = None, limit: int = 10):
    # Matches URLs like: /search?q=python&limit=5
    return {"query": q, "limit": limit}

4. Input Validation Layer: SQLModel Blueprints

One of FastAPI’s greatest strengths is its native integration with the validation ecosystem. With SQLModel, we can declare a model that serves a dual purpose: it acts as a database table entity and a request validation layout simultaneously.

from pydantic import BaseModel, Field

# Define the structural schema blueprint for incoming registration requests
class UserSignUpSchema(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    email: str
    age: int = Field(ge=18) # Age must be greater than or equal to 18
    premium_member: bool = False

5. Capturing Request Data (Body Payload vs. HTML Forms)

FastAPI is highly optimized for JSON API data exchange out of the box, but it can easily process classic HTML form submissions by installing the python-multipart module.

Here is how to capture both JSON and Form inputs side-by-side:

from fastapi import FastAPI, Form
from pydantic import BaseModel

app = FastAPI()

class StudentPayload(BaseModel):
    name: str
    course: str

# A. Capturing a raw JSON Payload (Standard API behavior)
@app.post("/students/json")
def add_student_json(payload: StudentPayload):
    return {"source": "JSON Body", "received_data": payload.dict()}

# B. Capturing traditional HTML Form Data (Web Form submission behavior)
@app.post("/students/form")
def add_student_form(name: str = Form(...), course: str = Form(...)):
    return {"source": "HTML Form", "name": name, "course": course}

6. Database Integrations with SQLModel Sessions

Now let’s establish our live database connection pipeline. We will initialize our SQLite engine, create our tables, and set up a clean dependency injection generator using a standard context Session block.

from fastapi import FastAPI, Depends, status
from sqlmodel import SQLModel, create_engine, Session, select, Field
from typing import List

class User(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    username: str = Field(min_length=3, max_length=20, index=True)
    email: str
    age: int = Field(ge=18)  # Age must be greater than or equal to 18
    premium_member: bool = False

app = FastAPI(title="FastAPI SQLModel Pipeline")

database_file = "sqlite:///database.db"
engine = create_engine(database_file, connect_args={"check_same_thread": False})

# Automatically initialize physical tables on application startup
SQLModel.metadata.create_all(engine)

# Dependency Injection Pattern to manage and yield database sessions safely
def get_db_session():
    with Session(engine) as db:
        yield db  # Closes the database transaction loop cleanly after request execution

# --- CORE WRITE ROUTE INTEGRATION ---
@app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED)
def create_user_record(user: User, db: Session = Depends(get_db_session)):
    db.add(user)
    db.commit()
    db.refresh(user)  # Populates the object with its new auto-generated database ID
    return user

# --- CORE READ ROUTE INTEGRATION ---
@app.get("/users", response_model=List[User])
def fetch_all_users(db: Session = Depends(get_db_session)):
    # Using SQLModel's custom, clean select statement syntax
    statement = select(User)
    users_list = db.exec(statement).all()
    return users_list

7. Session Cookies vs. Stateless JWT Architecture

For modern APIs, FastAPI shifts away from legacy stateful tracking to a stateless JWT (JSON Web Token) architecture.

Session Cookies (Stateful): The server is the “source of truth.” It creates a session in a database or cache (like Redis) and gives the client a random ID string (the cookie). Every request requires the server to look up this ID to see who the user is.

Stateless JWTs: The token itself is the “source of truth.” The server encrypts user data into a string, signs it with a secret key, and hands it to the client. The server doesn’t store anything; it just validates the signature on subsequent requests.

Why Use JWT Over Server Sessions?

  • Stateless Scaling: In a stateless setup, the server does not save any session tokens in its memory. Instead, once a user logs in, the server generates a signed cryptographic token called a JWT and hands it to the browser.
  • Self-Contained Data: The client attaches this token to the header of every subsequent request. The server simply verifies the token’s signature using its secret key. If the signature is valid, the request is instantly authorized, making it highly scalable across multi-server networks.

8. Implementing Security: Authentication & Authorization

Let’s build a secure authentication pipeline using JWT tokens. To make this work seamlessly, we will use FastAPI’s built-in OAuth2 security utilities (OAuth2PasswordBearer).

from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt
from jwt.exceptions import InvalidTokenError

app = FastAPI()

# Security Configurations
SECRET_KEY = "super_high_level_production_secret_key_salt"
ALGORITHM = "HS256"
TOKEN_EXPIRATION_MINUTES = 30

# Secure security gateway locator target configuration
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- TOKEN GENERATION UTILITY ---
def create_access_token(data: dict):
    payload_to_encode = data.copy()
    expire_time = datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRATION_MINUTES)
    payload_to_encode.update({"exp": expire_time})
    return jwt.encode(payload_to_encode, SECRET_KEY, algorithm=ALGORITHM)

# --- MIDDLEWARE AUTHENTICATION GATE DEPENDENCY ---
async def get_current_user_gate(token: str = Depends(oauth2_scheme)):
    try:
        # Decode the cryptographically signed JWT token payload
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        role: str = payload.get("role")
        if username is None:
            raise Exception("Login Error", "Please Enter Valid Username Or Password.")            
        return {"username": username, "role": role}
    except InvalidTokenError:  # Replaced JWTError with PyJWT's exception
        raise Exception("Login Error", "Please Enter Valid Username Or Password.")

# --- ROUTES ---
@app.post("/token")
def authenticate_user(form_data: OAuth2PasswordRequestForm = Depends()):
    # Simulated database validation check
    if form_data.username == "vipin" and form_data.password == "secure123":
        # Generate token payload containing identity markers and authorization roles
        token = create_access_token(data={"sub": form_data.username, "role": "admin"})
        return {"access_token": token, "token_type": "bearer"}

    raise HTTPException(status_code=400, detail="Incorrect username or password configuration.")

@app.get("/admin/panel")
def view_admin_panel(current_user: dict = Depends(get_current_user_gate)):
    # Authorization Access Check
    if current_user["role"] != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, 
            detail="Access Denied: You do not have sufficient authorization privileges."
        )        
    return {"message": f"Welcome back Administrator {current_user['username']}! Opening control systems..."}

9. Automatic Interactive Documentation (Swagger UI)

One of the most useful out-of-the-box features of FastAPI is its automatic interactive documentation.

The moment you declare your SQLModel models and add type hints to your routes, FastAPI automatically generates a complete openAPI documentation portal for your project.

How to Access the Docs Portal

While your Uvicorn server is running locally, open your web browser and navigate to:

  • Swagger UI Documentation: http://127.0.0.1:8000/docs (An interactive testing dashboard where you can run live test queries directly from your browser).
  • ReDoc Documentation: http://127.0.0.1:8000/redoc (A beautifully clean, structured corporate documentation layout).