Using JWT in FastAPI with PostgreSQL Integration

Photo by Casey Allen on Unsplash

JSON Web Token (JWT) is a secure and efficient way to manage user authentication and authorization. When paired with a PostgreSQL database, you get a robust system for handling user credentials, session management, and access control. This article demonstrates how to implement JWT-based authentication in FastAPI and connect the application to a PostgreSQL database.

Overview

This FastAPI application provides login an signup endpoints for user authentication. Passwords are securely hashed, credentials are stored in PostgreSQL, and JWT access tokens are issued to authenticated users.

We will cover:

1. Connecting to PostgreSQL with SQLAlchemy.
2. Setting up JWT-based authentication.

1. Connecting to PostgreSQL

The application uses SQLAlchemy to connect to a PostgreSQL database. The connection details are specified in the session.py file.

Code: Database Connection

from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from app.core.config import settings

Base: DeclarativeMeta = declarative_base()
# Configure the engine with PostgreSQL connection details
engine = create_engine(
    "postgresql+psycopg2://[username]:[password]@[host:port]/[database]"
)
# Create a session factory for interacting with the database
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Dependency to provide a database session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Explanation

• Engine: The create_engine function initializes the connection to PostgreSQL using the psycopg2 driver.
• SessionLocal: Manages interactions with the database, ensuring proper cleanup of resources.
• Dependency Injection: The get_db function ensures that database sessions are passed safely to endpoint handlers.

Environment Configuration

Sensitive details like SECRET_KEY and database credentials should be stored in an .env file:

SECRET_KEY=your-secret-key
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
DATABASE_URL=postgresql+psycopg2://postgres:mysecretpassword@localhost/immi_bot

To load these settings, you can use python-decouple or dotenv.

2. JWT-Based Authentication

Router: Authentication Endpoints

The auth router handles user login and signup requests, issuing JWTs upon successful authentication or registration.

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.schemas.auth import LoginRequest, LoginResponse, SignupRequest, SignupResponse
from app.db.session import get_db
from app.services.auth_service import authenticate_user, create_user
from app.db.models import User
from app.core.security import create_access_token
 
router = APIRouter(prefix="/auth", tags=["Authentication"])

@router.post("/login", response_model=LoginResponse)
def login(request: LoginRequest, db: Session = Depends(get_db)):
    """
    Login route for obtaining an access token after verifying credentials.
    """
    user = authenticate_user(db, request.email, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    # Generate an access token
    access_token = create_access_token({"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}
@router.post("/signup", response_model=SignupResponse)

def signup(request: SignupRequest, db: Session = Depends(get_db)):
    """
    Sign-up route for creating a new user and storing their credentials securely.
    """
    user = db.query(User).filter(User.email == request.email).first()
    if user:
        raise HTTPException(status_code=400, detail="Email already registered")
    # Create the new user
    new_user = create_user(db, request.email, request.password)
    
    # Generate access token for the new user
    access_token = create_access_token({"sub": new_user.email})
    return {"access_token": access_token, "token_type": "bearer"}

Key Highlights

1. Dependencies: The get_db dependency injects a database session.
2. Error Handling: HTTP exceptions are raised for invalid credentials or duplicate registrations.
3. JWT Generation: The create_access_token function generates a token for each user.

Authentication Logic

The authentication logic resides in auth_service.py and uses secure hashing for passwords.

from sqlalchemy.orm import Session
from app.db.models import User
from app.core.security import hash_password, verify_password

def authenticate_user(db: Session, email: str, password: str) -> User:
    """
    Authenticate a user based on email and password.
    Returns the user if authentication is successful, otherwise None.
    """
    user = db.query(User).filter(User.email == email).first()
    if user and verify_password(password, user.password_hash):
        return user
    return None

def create_user(db: Session, email: str, password: str) -> User:
    """
    Create a new user in the database with a hashed password.
    """
    hashed_password = hash_password(password)
    new_user = User(email=email, password_hash=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

JWT Token Creation

The security.py file handles password hashing and JWT creation.

from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import JWTError, jwt
from app.core.config import settings

# CryptContext for hashing and verifying passwords
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

SECRET_KEY = settings.SECRET_KEY
ALGORITHM = settings.ALGORITHM
ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """
    Verify if the given password matches the hashed password.
    """
    return pwd_context.verify(plain_password, hashed_password)

def hash_password(password: str) -> str:
    """
    Hash a password before storing it in the database.
    """
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) -> str:
    """
    Create a JWT access token with the given data and expiration time.
    """
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Putting It All Together

The main.py file initializes the FastAPI application, loads routes, and starts the server.

from fastapi import FastAPI
import uvicorn      
from app.routers.auth import router as auth_router

# Create FastAPI app
app = FastAPI(title="Authentication with JWT and PostgreSQL", version="1.0.0")
# Include routers

app.include_router(auth_router, tags=["Authentication"])

@app.get("/")
async def root():
    """Root endpoint for testing the server."""
    return {"message": "Welcome to the JWT Authentication API!"}
# Entry point to run the app

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

Conclusion

By combining JWT for authentication and PostgreSQL for data persistence, this FastAPI application demonstrates a secure and scalable solution for user management. The modular approach ensures easy maintenance and adaptability for future enhancements.