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
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 venvvenv\Scripts\activate
source venv/bin/activate # On Windows, use
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() 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) 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 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 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 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:
- create a new user
curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"username": "testuser", "email": "[email protected]", "password": "testpassword"}'
- Obtain a JWT token:
curl -X POST "http://127.0.0.1:8000/token" -d "username=testuser&password=testpassword"
- 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.
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.