Setup FastAPI email verification and password reset functionality with Redis and PostgreSQL

Update system packages

Start by updating your package manager and installing required system dependencies.

sudo apt update && sudo apt upgrade -y
sudo apt install -y python3 python3-pip python3-venv redis-server postgresql postgresql-contrib

sudo dnf update -y
sudo dnf install -y python3 python3-pip redis postgresql postgresql-server postgresql-contrib

Configure PostgreSQL database

Initialize PostgreSQL and create a database and user for your FastAPI application.

sudo systemctl enable --now postgresql

sudo postgresql-setup --initdb
sudo systemctl enable --now postgresql

sudo -u postgres createuser --interactive
# Enter username: fastapi_user
# Shall the new role be a superuser? n
# Shall the new role be allowed to create databases? y
# Shall the new role be allowed to create more new roles? n

sudo -u postgres createdb fastapi_db -O fastapi_user

sudo -u postgres psql -c "ALTER USER fastapi_user PASSWORD 'your_secure_password';"

Configure Redis server

Enable Redis and configure it for session storage with proper security settings.

sudo systemctl enable --now redis-server

# Bind to localhost only for security
bind 127.0.0.1

# Set a password (uncomment and change)
requirepass your_redis_password

# Configure memory policy
maxmemory 256mb
maxmemory-policy allkeys-lru

# Enable persistence
save 900 1
save 300 10
save 60 10000

sudo systemctl restart redis-server

Create Python virtual environment

Set up an isolated Python environment for your FastAPI application dependencies.

mkdir fastapi-auth && cd fastapi-auth
python3 -m venv venv
source venv/bin/activate

Install FastAPI dependencies

Install all required Python packages for authentication, database access, and email functionality.

pip install fastapi[all] uvicorn sqlalchemy psycopg2-binary redis python-jose[cryptography] passlib[bcrypt] python-multipart email-validator jinja2 aiosmtplib slowapi

Create database models

Define the user model and database schema with SQLAlchemy for PostgreSQL integration.

from sqlalchemy import Column, Integer, String, Boolean, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import os

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=False)
    is_verified = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class PasswordResetToken(Base):
    __tablename__ = "password_reset_tokens"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, nullable=False)
    token = Column(String, unique=True, nullable=False)
    expires_at = Column(DateTime, nullable=False)
    used = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)

class EmailVerificationToken(Base):
    __tablename__ = "email_verification_tokens"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, nullable=False)
    token = Column(String, unique=True, nullable=False)
    expires_at = Column(DateTime, nullable=False)
    used = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# Database connection
DATABASE_URL = "postgresql://fastapi_user:your_secure_password@localhost/fastapi_db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Create tables
Base.metadata.create_all(bind=engine)

Configure authentication utilities

Set up password hashing, JWT tokens, and Redis session management.

from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
import secrets
import redis
import json

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT settings
SECRET_KEY = "your-secret-key-change-this-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Redis connection
redis_client = redis.Redis(host='localhost', port=6379, password='your_redis_password', decode_responses=True)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: 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

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            return None
        return email
    except JWTError:
        return None

def generate_verification_token() -> str:
    return secrets.token_urlsafe(32)

def store_session(session_id: str, user_data: dict, expire_seconds: int = 1800):
    redis_client.setex(session_id, expire_seconds, json.dumps(user_data))

def get_session(session_id: str) -> dict:
    session_data = redis_client.get(session_id)
    if session_data:
        return json.loads(session_data)
    return None

def delete_session(session_id: str):
    redis_client.delete(session_id)

Configure email functionality

Set up SMTP email sending for verification and password reset emails.

import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from jinja2 import Template
import os

# SMTP configuration
SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
SMTP_USERNAME = os.getenv("SMTP_USERNAME", "your-email@example.com")
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "your-app-password")
FROM_EMAIL = os.getenv("FROM_EMAIL", "noreply@example.com")

# Email verification template
VERIFICATION_EMAIL_TEMPLATE = """



    


    <h2>Email Verification Required</h2>
    <p>Hello,</p>
    <p>Please click the link below to verify your email address:</p>
    <p><a href="{{%20verification_url%20}}">Verify Email</a></p>
    <p>This link will expire in 24 hours.</p>
    <p>If you didn't create an account, please ignore this email.</p>


"""

# Password reset template
PASSWORD_RESET_TEMPLATE = """



    


    <h2>Password Reset Request</h2>
    <p>Hello,</p>
    <p>You requested a password reset. Click the link below to reset your password:</p>
    <p><a href="{{%20reset_url%20}}">Reset Password</a></p>
    <p>This link will expire in 1 hour.</p>
    <p>If you didn't request this, please ignore this email.</p>


"""

async def send_email(to_email: str, subject: str, html_content: str):
    message = MIMEMultipart("alternative")
    message["Subject"] = subject
    message["From"] = FROM_EMAIL
    message["To"] = to_email
    
    html_part = MIMEText(html_content, "html")
    message.attach(html_part)
    
    try:
        await aiosmtplib.send(
            message,
            hostname=SMTP_HOST,
            port=SMTP_PORT,
            username=SMTP_USERNAME,
            password=SMTP_PASSWORD,
            use_tls=True,
        )
        return True
    except Exception as e:
        print(f"Failed to send email: {e}")
        return False

async def send_verification_email(email: str, token: str):
    verification_url = f"http://localhost:8000/verify-email?token={token}"
    template = Template(VERIFICATION_EMAIL_TEMPLATE)
    html_content = template.render(verification_url=verification_url)
    
    return await send_email(
        to_email=email,
        subject="Verify Your Email Address",
        html_content=html_content
    )

async def send_password_reset_email(email: str, token: str):
    reset_url = f"http://localhost:8000/reset-password?token={token}"
    template = Template(PASSWORD_RESET_TEMPLATE)
    html_content = template.render(reset_url=reset_url)
    
    return await send_email(
        to_email=email,
        subject="Reset Your Password",
        html_content=html_content
    )

Create main FastAPI application

Build the main application with all authentication endpoints and security middleware.

from fastapi import FastAPI, HTTPException, Depends, Request, Response
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from pydantic import BaseModel, EmailStr
import uuid

from models import User, PasswordResetToken, EmailVerificationToken, SessionLocal
from auth_utils import (
    verify_password, get_password_hash, create_access_token, 
    verify_token, generate_verification_token, store_session, 
    get_session, delete_session
)
from email_service import send_verification_email, send_password_reset_email

# Initialize FastAPI app
app = FastAPI(title="FastAPI Authentication API", version="1.0.0")

# Rate limiting
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Add your frontend URL
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Security
security = HTTPBearer()

# Pydantic models
class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserLogin(BaseModel):
    email: EmailStr
    password: str

class PasswordReset(BaseModel):
    email: EmailStr

class PasswordResetConfirm(BaseModel):
    token: str
    new_password: str

class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool
    is_verified: bool

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

# Authentication dependency
def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db)):
    token = credentials.credentials
    email = verify_token(token)
    if email is None:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")
    
    user = db.query(User).filter(User.email == email).first()
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    
    return user

@app.post("/register", response_model=dict)
@limiter.limit("5/minute")
async def register_user(request: Request, user_data: UserCreate, db: Session = Depends(get_db)):
    # Check if user already exists
    existing_user = db.query(User).filter(User.email == user_data.email).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # Create new user
    hashed_password = get_password_hash(user_data.password)
    user = User(
        email=user_data.email,
        hashed_password=hashed_password,
        is_active=True,
        is_verified=False
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    
    # Generate verification token
    verification_token = generate_verification_token()
    expires_at = datetime.utcnow() + timedelta(hours=24)
    
    token_record = EmailVerificationToken(
        email=user_data.email,
        token=verification_token,
        expires_at=expires_at
    )
    db.add(token_record)
    db.commit()
    
    # Send verification email
    email_sent = await send_verification_email(user_data.email, verification_token)
    
    return {
        "message": "User registered successfully",
        "email_sent": email_sent,
        "user_id": user.id
    }

@app.post("/login", response_model=dict)
@limiter.limit("10/minute")
async def login(request: Request, response: Response, user_credentials: UserLogin, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == user_credentials.email).first()
    
    if not user or not verify_password(user_credentials.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect email or password")
    
    if not user.is_active:
        raise HTTPException(status_code=401, detail="Account is deactivated")
    
    if not user.is_verified:
        raise HTTPException(status_code=401, detail="Please verify your email address")
    
    # Create access token
    access_token_expires = timedelta(minutes=30)
    access_token = create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )
    
    # Create session
    session_id = str(uuid.uuid4())
    session_data = {
        "user_id": user.id,
        "email": user.email,
        "is_verified": user.is_verified
    }
    store_session(session_id, session_data, expire_seconds=1800)
    
    # Set session cookie
    response.set_cookie(
        key="session_id",
        value=session_id,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=1800
    )
    
    return {
        "access_token": access_token,
        "token_type": "bearer",
        "user": {
            "id": user.id,
            "email": user.email,
            "is_verified": user.is_verified
        }
    }

@app.get("/verify-email")
@limiter.limit("5/minute")
async def verify_email(request: Request, token: str, db: Session = Depends(get_db)):
    token_record = db.query(EmailVerificationToken).filter(
        EmailVerificationToken.token == token,
        EmailVerificationToken.used == False,
        EmailVerificationToken.expires_at > datetime.utcnow()
    ).first()
    
    if not token_record:
        raise HTTPException(status_code=400, detail="Invalid or expired verification token")
    
    # Update user verification status
    user = db.query(User).filter(User.email == token_record.email).first()
    if user:
        user.is_verified = True
        token_record.used = True
        db.commit()
        
        return {"message": "Email verified successfully"}
    
    raise HTTPException(status_code=404, detail="User not found")

@app.post("/request-password-reset")
@limiter.limit("3/minute")
async def request_password_reset(request: Request, reset_data: PasswordReset, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == reset_data.email).first()
    
    if not user:
        # Don't reveal if email exists
        return {"message": "If the email exists, a reset link has been sent"}
    
    # Generate reset token
    reset_token = generate_verification_token()
    expires_at = datetime.utcnow() + timedelta(hours=1)
    
    token_record = PasswordResetToken(
        email=reset_data.email,
        token=reset_token,
        expires_at=expires_at
    )
    db.add(token_record)
    db.commit()
    
    # Send reset email
    await send_password_reset_email(reset_data.email, reset_token)
    
    return {"message": "If the email exists, a reset link has been sent"}

@app.post("/reset-password")
@limiter.limit("5/minute")
async def reset_password(request: Request, reset_data: PasswordResetConfirm, db: Session = Depends(get_db)):
    token_record = db.query(PasswordResetToken).filter(
        PasswordResetToken.token == reset_data.token,
        PasswordResetToken.used == False,
        PasswordResetToken.expires_at > datetime.utcnow()
    ).first()
    
    if not token_record:
        raise HTTPException(status_code=400, detail="Invalid or expired reset token")
    
    # Update user password
    user = db.query(User).filter(User.email == token_record.email).first()
    if user:
        user.hashed_password = get_password_hash(reset_data.new_password)
        token_record.used = True
        db.commit()
        
        return {"message": "Password reset successfully"}
    
    raise HTTPException(status_code=404, detail="User not found")

@app.get("/me", response_model=UserResponse)
async def get_current_user_info(current_user: User = Depends(get_current_user)):
    return UserResponse(
        id=current_user.id,
        email&a