🔐 FastAPI JWT Authentication with SQLite Database for Secure APIs

Building robust and secure APIs is a cornerstone of modern web development. FastAPI, with its incredible speed and developer-friendly features, is a fantastic choice for this. When it comes to securing your API, JSON Web Tokens (JWT) offer a stateless and efficient authentication mechanism.

In this blog post, we’ll walk through building a FastAPI application that integrates JWT-based authentication with a SQLite database for user management. We’ll cover everything from setting up your project, defining database models, implementing authentication logic, and securing your API endpoints.

1. Project Setup and Dependencies

First, let’s get our project environment ready. We’ll use pip and a pyproject.toml file to manage our dependencies.

Your pyproject.toml specifies all the necessary libraries:

# pyproject.toml
[project]
name = "fastpi-jwt-auth"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.116.1",
    "passlib[bcrypt]>=1.7.4",
    "pydantic>=2.11.7",
    "python-jose[cryptography]>=3.5.0",
    "python-multipart>=0.0.20",
    "sqlalchemy>=2.0.41",
    "uvicorn>=0.35.0",
]

To install these, navigate to your project directory in the terminal and run:

pip install -e .

This command installs all dependencies listed in your pyproject.toml.

2. Database Configuration (database.py)

We’ll use SQLite as our database for simplicity, which is perfect for development. SQLAlchemy is our chosen Object-Relational Mapper (ORM) to interact with the database.

Your database.py file sets up the connection:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# Define the URL for our SQLite database
# sqlite:///./todosapp.db means a file named todosapp.db in the current directory
DATABASE_URL = "sqlite:///./todosapp.db"
# Create the SQLAlchemy engine
# connect_args={"check_same_thread": False} is crucial for SQLite
# It tells SQLite that multiple threads might access the database connection,
# which is common in web servers like Uvicorn.
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
# Create a SessionLocal class for database sessions
# autocommit=False ensures transactions are managed manually
# autoflush=False prevents immediate flushing of changes
# bind=engine links sessions to our database engine
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class for our declarative models
# All SQLAlchemy models will inherit from this Base
Base = declarative_base()

Explanation:

• DATABASE_URL: This specifies the path to your SQLite database file.
• create_engine: This function creates a SQLAlchemy engine, which is responsible for communication with the database. The check_same_thread=False argument is essential for SQLite when used with multi-threaded environments like FastAPI's Uvicorn server, as SQLite typically expects to be accessed by only one thread.
• sessionmaker: This creates a factory for new Session objects. Sessions are the primary way to interact with the database (add, query, update, delete).
• declarative_base: This function returns a base class for declarative model definitions. Your SQLAlchemy models will inherit from this Base.

3. Database Models (models.py)

Next, we define our User model, which maps to a table in our SQLite database.

# models.py
from sqlalchemy import Column, String, Integer
from database import Base # Import Base from our database.py

class User(Base):
    # Define the table name in the database
    __tablename__ = 'users'
    # Define columns for the 'users' table
    id = Column(Integer, primary_key=True, index=True) # Primary key, auto-incrementing, indexed for fast lookups
    username = Column(String, unique=True) # Unique username
    password = Column(String) # Hashed password

Explanation:

• We import Column, String, and Integer from SQLAlchemy to define column types.
• User inherits from Base, linking it to our SQLAlchemy declarative system.
• __tablename__ = 'users' explicitly sets the table name in the database.
• id: An integer primary key that will be automatically managed by the database. index=True creates a database index on this column for faster queries.
• username: A string column for the username, set to unique=True to ensure no two users have the same username.
• password: A string column to store the hashed password. Never store plain pass words!

4. Authentication Logic (auth.py)

This file is the heart of our authentication system, handling user creation, login, token generation, and token validation.

# auth.py
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter, HTTPException, Depends
from typing import Annotated
from pydantic import BaseModel
from sqlalchemy.orm import Session
from starlette import status
from database import SessionLocal
from models import User
from passlib.context import CryptContext # For password hashing
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer # For OAuth2 flow
from jose import JWTError, jwt # For JWT encoding/decoding

router = APIRouter(prefix="/auth", tags=["auth"])
# Secret key for signing JWTs - **CHANGE THIS IN PRODUCTION!**
SECRET_KEY = "0553abd6dc9e05eae3f5f0ae457d47ccbf9a10b90bfc9cba896c643cdf912c2cdf702314aa41a33a0cdcf294547eda3a"
ALGORITHM = "HS256" # Hashing algorithm for JWT
# Password hashing context (using bcrypt)
dcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# OAuth2PasswordBearer for handling token extraction from headers
oauth2_brearer = OAuth2PasswordBearer(tokenUrl="auth/token")
# Pydantic model for user registration request body
class CreateUserRequest(BaseModel):
    username: str
    password: str
# Pydantic model for token response
class Token(BaseModel):
    access_token: str
    token_type: str
# Dependency to get a database session
def get_db():
    db = SessionLocal() # Create a new session
    try:
        yield db # Yield the session to the FastAPI route
    finally:
        db.close() # Ensure the session is closed after the request
# Annotated dependency for easy type hinting in routes
db_dependancy = Annotated[Session, Depends(get_db)]
# --- API Endpoints ---
@router.post("/", status_code=status.HTTP_201_CREATED )
async def create_user(db: db_dependancy, user: CreateUserRequest):
    # Hash the password before storing it
    create_user_model = User(
        username=user.username,
        password=dcrypt_context.hash(user.password))
    db.add(create_user_model) # Add the new user to the session
    db.commit() # Commit the transaction to save to DB
    return {"message": "User created successfully"} # Return a success message
@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: db_dependancy):
    # Authenticate user credentials
    user = authenticate_user(form_data.username, form_data.password, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"} # Standard header for auth challenges
        )
    # Create an access token for the authenticated user
    token = create_access_token(user.username, user.id, timedelta(minutes=30))
    return {"access_token": token, "token_type": "bearer"}
# --- Helper Functions ---
def authenticate_user(username:str, password: str, db):
    # Query database for user
    user = db.query(User).filter(User.username == username).first()
    if not user:
        return False # User not found
    # Verify provided password against hashed password
    if not dcrypt_context.verify(password, user.password):
        return False # Password mismatch
    return user # User authenticated
def create_access_token(username:str, user_id : int, expires_delta: timedelta | None = None):
    # Payload for the JWT
    to_encode = {"sub": username, "id": user_id} # 'sub' is standard for subject
    # Set token expiration time
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15) # Default 15 min expiry
    to_encode.update({"exp": expire}) # Add expiration to payload
    # Encode the JWT using the secret key and algorithm
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def get_current_user(token: Annotated[str, Depends(oauth2_brearer)]):
    try:
        # Decode the token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        user_id = payload.get("id")
        if username is None or user_id is None:
            # If payload is incomplete
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials",
                headers={"WWW-Authenticate": "Bearer"}
            )
        return {"username": username, "id": user_id} # Return user info from token
    except JWTError:
        # If token is invalid or expired
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"}
        )

Explanation of auth.py:

• APIRouter: Organizes our authentication-related endpoints under the /auth prefix.
• SECRET_KEY, ALGORITHM: Essential for JWT signing and verification. Always use a strong, randomly generated key in production and keep it secret!
• CryptContext (passlib): Used to securely hash and verify user passwords (using bcrypt in this case). This is critical for security.
• OAuth2PasswordBearer (fastapi.security): This class handles extracting the token from the Authorization: Bearer header of incoming requests. tokenUrl points to our login endpoint.
• CreateUserRequest, Token (Pydantic Models): Define the expected structure for incoming request bodies (for user creation) and outgoing response bodies (for the token).
• get_db() Dependency: This is a core FastAPI pattern. It provides a database session (SessionLocal) for each request and ensures the session is properly closed afterward, preventing resource leaks. The yield keyword makes it a dependency.
• db_dependancy: An Annotated type hint that simplifies injecting the database session into our route functions.

@router.post("/") (Create User):

• Takes CreateUserRequest as input for username and password.
• Hashes the password using dcrypt_context.hash().
• Creates a new User object, adds it to the database session, and commits the transaction.

@router.post("/token") (Login):

• Uses OAuth2PasswordRequestForm to expect username and password as form data (standard for OAuth2 token endpoints).
• Calls authenticate_user to verify credentials. If invalid, raises HTTPException(401 UNAUTHORIZED).
• If authenticated, calls create_access_token to generate a JWT.
• Returns the JWT and token type.
• authenticate_user(): Queries the database for the user and verifies the password using dcrypt_context.verify().

create_access_token():

• Constructs the JWT payload including sub (subject, here username) and id (user ID).
• Sets an exp (expiration) timestamp for the token. JWTs are usually short-lived.
• Encodes the payload using jwt.encode() with the SECRET_KEY and ALGORITHM.

get_current_user() Dependency:

• This function is itself a dependency. It uses oauth2_brearer to get the token from the request.
• It then decodes and validates the JWT using jwt.decode().
• If the token is invalid, expired, or missing essential data, it raises an HTTPException(401 UNAUTHORIZED).
• If successful, it returns a dictionary containing the authenticated user’s username and id.

5. Main Application (main.py)

This is our main FastAPI application file, where we tie everything together.

# main.py
from fastapi import FastAPI, HTTPException, status, Depends
import auth # Import our authentication module
import models # Import our models module
from database import engine, SessionLocal # Import engine and SessionLocal from database.py
from sqlalchemy.orm import Session
from typing import Annotated
from auth import get_current_user # Import the get_current_user dependency

app = FastAPI()
# Include the authentication router, making its endpoints available
app.include_router(auth.router)
# Create database tables defined in models.py when the application starts
# This will create the 'users' table if it doesn't already exist
models.Base.metadata.create_all(bind=engine)
# This get_db is duplicated in auth.py, but it's good practice to have it
# readily available in main.py if other parts of the app need a DB session.
# For this specific setup, the one in auth.py is sufficient for auth routes.
# However, for new routes in main.py, this would be used.
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()
# Annotated dependencies for easy use
db_dependancy = Annotated[Session, Depends(get_db)] # Database session dependency
user_dependancy = Annotated[dict, Depends(get_current_user)] # Current authenticated user dependency
# An example protected endpoint
@app.get('/', status_code=status.HTTP_200_OK)
async def user(user: user_dependancy, db: db_dependancy):
    # The get_current_user dependency already handles authentication
    # If get_current_user raises an HTTPException, this code won't be reached.
    # So, a simple check if user is None here is redundant if get_current_user is robust.
    # We can directly use the 'user' dictionary.
    return {"message": f"Hello, {user['username']} with ID: {user['id']}!"}

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

@app.get('/') (Protected Endpoint):

• This is an example of a route that requires authentication.
• By adding user: user_dependancy, FastAPI will automatically execute get_current_user. If get_current_user successfully returns user data (meaning the token is valid), that data will be passed to the user parameter. If get_current_user raises an HTTPException, FastAPI will return that error response directly, and the code inside async def user will not execute.
• It now returns a personalized greeting using the extracted username and id from the token payload.

• if name == "main":: This standard Python construct allows you to run your application directly using uvicorn.
  Install dependencies:

• pip install -e .
  Run the application:

• python main.py

• You should see output indicating Uvicorn is running, typically on http://0.0.0.0:8000.

• Access FastAPI Docs: Open your browser and go to http://127.0.0.1:8000/docs (or http://localhost:8000/docs). You'll see the interactive API documentation (Swagger UI).

Testing the Endpoints:

1. Create a User (Sign Up)

• Go to /docs.
• Find the POST /auth/ endpoint.
• Click “Try it out”.
• In the “Request body” field, enter a username and password:
• { "username": "testuser", "password": "testpassword" }
• Click “Execute”. You should get a 201 Created response. This will create the todosapp.db file in your project directory.

2. Log In and Get a Token

• Find the POST /auth/token endpoint.
• Click “Try it out”.
• For username and password fields, enter the credentials you just created (testuser, testpassword).
• Click “Execute”.
• You should receive a 200 OK response with an access_token and token_type: "bearer". Copy the access_token value.

3. Access a Protected Endpoint

• At the top right of the Swagger UI, click the “Authorize” button (or the lock icon next to the endpoints).
• In the dialog box, under “BearerAuth”, paste your copied access_token into the Value field (without the "Bearer " prefix, just the token itself).
• Click “Authorize”, then “Close”.
• Now, try the GET / endpoint.
• Click “Try it out” and then “Execute”.
• You should receive a 200 OK response with a message like: {"message": "Hello, {'username': 'testuser', 'id': 1}!"}.
• If you try to access the GET / endpoint without authorization (by clicking "Unauthorize" first, or not providing a token), you will get a 401 Unauthorized error. If want to try the Basic Authentication — have a look at this
  

  Implementing Basic Authentication in FastAPI

  Gopinath V ・ Jul 15

  #webdev #python #api #fastapi