DEV Community

Rishi Sharma
Rishi Sharma

Posted on

Building JWT Auth Chaining with FastAPI and Python

In this article, we'll explore how to implement JWT (JSON Web Token) authentication in a FastAPI application. We'll cover everything from setting up the project to securing endpoints with JWT. By the end, you'll have a fully functional authentication system with chained JWT verification.

  • JSON Web Token
  • Setting Up FastAPI Project
  • Creating User Models and Database
  • Implementing JWT Authentication
  • Securing Endpoints
  • Testing the Authentication System

JWT Flow

JSON Web Token ( JWT )

JWT is a compact, URL-safe token format that is commonly used for securely transmitting information between parties. JWT tokens are often used for authentication and authorization purposes in web applications.

Setting Up FastAPI Project

First, let's set up a new FastAPI project. Ensure you have Python installed and then create a virtual environment:

python -m venv venv
source venv/bin/activate # On Windows, use
venv\Scripts\activate

Next, install FastAPI and Uvicorn (an ASGI server):

pip install fastapi uvicorn[standard] python-jose passlib[bcrypt] pydantic sqlalchemy

Create a new directory for your project and add the following files:

my_fastapi_app/

├── main.py
├── models.py
├── database.py
├── schemas.py
└── auth.py

Creating User Models and Database

Let's start by defining our database models and setting up the database connection. We'll use SQLAlchemy for ORM.

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

Enter fullscreen mode Exit fullscreen mode

models.py

from sqlalchemy import Column, Integer, String, Boolean
from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

Enter fullscreen mode Exit fullscreen mode

schemas.py

from pydantic import BaseModel
from typing import Optional

class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True

Enter fullscreen mode Exit fullscreen mode

Implementing JWT

We need to implement JWT token creation and verification. We'll use the python-jose library for handling JWTs.

auth.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from . import models, schemas, database

# Secret key to encode JWT tokens
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def authenticate_user(db: Session, username: str, password: str):
    user = db.query(models.User).filter(models.User.username == username).first()
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(database.get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = schemas.TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = db.query(models.User).filter(models.User.username == token_data.username).first()
    if user is None:
        raise credentials_exception
    return user

Enter fullscreen mode Exit fullscreen mode

Securing end points

Now, let's create our FastAPI application and secure endpoints with JWT authentication.

main.py

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import models, schemas, database, auth

app = FastAPI()

models.Base.metadata.create_all(bind=database.engine)

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(database.get_db)):
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    hashed_password = auth.get_password_hash(user.password)
    db_user = models.User(username=user.username, email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.post("/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(database.get_db), form_data: OAuth2PasswordRequestForm = Depends()):
    user = auth.authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = auth.create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(current_user: schemas.User = Depends(auth.get_current_user)):
    return current_user

Enter fullscreen mode Exit fullscreen mode

Testing the Authentication System

To test our authentication system, we need to run the FastAPI application and use an HTTP client like curl or Postman to interact with our endpoints.

Run the application:

uvicorn main:app --reload

Now, let's test our endpoints using curl:

  1. create a new user

curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"username": "testuser", "email": "test@example.com", "password": "testpassword"}'

  1. Obtain a JWT token:

curl -X POST "http://127.0.0.1:8000/token" -d "username=testuser&password=testpassword"

  1. Access a protected endpoint:

curl -X GET "http://127.0.0.1:8000/users/me/" -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Replace YOUR_ACCESS_TOKEN with the token obtained in step 2.

Conclusion

In this article, we've built a complete JWT authentication system using FastAPI and Python. We've covered the basics of JWT, set up a FastAPI project, created user models and a database, implemented JWT authentication, secured endpoints, and tested our system. This setup provides a robust foundation for any application requiring user authentication and authorization.

JWT Structure

Feel free to expand upon this system by adding more features such as role-based access control, refresh tokens, and more advanced user management. Happy coding!

If you found this article helpful, please leave a comment or reach out with any questions.

Top comments (0)

Some comments may only be visible to logged-in visitors. Sign in to view all comments.