Leapcell

Posted on Oct 2

Build a Perfect Blog with FastAPI: Add User System

#webdev #programming #python #backend

In the previous article, we used FastAPI to build a basic personal blog and successfully deployed it.

However, this blog has a serious security issue: anyone can create articles at will.

In the following tutorial, we will add a user and authentication system to this blog to make it more secure.

Without further ado, let's get started.

Introduction to Authentication Methods

In web development, the two most common authentication methods are Token-based (e.g., JWT) and Session-based (Cookie).

JWT (JSON Web Tokens): This is currently the most popular authentication method. After a user logs in, the server generates a token and returns it to the client. The client includes this token in subsequent requests, and the server only needs to validate the token to confirm the user's identity. Since the server doesn't need to store user state, this method is very suitable for distributed, horizontally scalable large-scale applications.
Session-Cookie: After a user logs in, the server creates a session and returns the Session ID to the browser via a cookie. The browser automatically includes this cookie in subsequent requests. The server then identifies the user by looking up the corresponding session information based on the Session ID.

In this tutorial, we will choose the traditional Session-Cookie method. Because our blog is a monolith with a simple architecture, using Session-Cookies for authentication is the most direct, classic, and sufficiently secure approach.

Step 1: Create the User Module

Before we can handle authentication, we first need a user system.

1. Create the User Data Model

Open the models.py file and add the User model above the Post class.

# models.py
import uuid
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str = Field(unique=True, index=True)
    password: str # The stored password will be the encrypted hash

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)

Since we configured the create_db_and_tables function in main.py in the first article, it will automatically detect all SQLModel models and create the corresponding database tables when the application starts. Therefore, we don't need to execute SQL statements manually.

If you need to execute SQL manually and your database was created on Leapcell,

you can easily execute SQL statements using the graphical interface. Just go to the Database management page on the website, paste the above statements into the SQL interface, and execute them.

2. Install the Password Encryption Library

For security, user passwords must never be stored in the database in plaintext. We will use the bcrypt library to hash the passwords.

First, add bcrypt to your requirements.txt file:

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt

Then, run the installation command:

pip install -r requirements.txt

Step 2: Implement User Registration and Validation Logic

Next, we will create functions to handle user data and validate passwords.

Create a new file users_service.py in the project's root directory to store user-related business logic.

# users_service.py
import bcrypt
from sqlmodel import Session, select
from models import User

def get_user_by_username(username: str, session: Session) -> User | None:
    """Find a user by username"""
    statement = select(User).where(User.username == username)
    return session.exec(statement).first()

def create_user(user_data: dict, session: Session) -> User:
    """Create a new user and hash the password"""
    # Convert the plaintext password to bytes
    password_bytes = user_data["password"].encode('utf-8')
    # Generate a salt and hash the password
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(password_bytes, salt)

    new_user = User(
        username=user_data["username"],
        # Decode the hashed password (bytes) into a string for database storage
        password=hashed_password.decode('utf-8')
    )

    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

Then, create an auth_service.py file to handle user authentication.

# auth_service.py
import bcrypt
from sqlmodel import Session
from models import User
from users_service import get_user_by_username

def validate_user(username: str, plain_password: str, session: Session) -> User | None:
    """Validate if the username and password match"""
    user = get_user_by_username(username, session)
    if not user:
        return None

    # Encode both the input plaintext password and the stored hashed password to bytes
    plain_password_bytes = plain_password.encode('utf-8')
    hashed_password_bytes = user.password.encode('utf-8')

    # Use bcrypt.checkpw for comparison
    if bcrypt.checkpw(plain_password_bytes, hashed_password_bytes):
        return user # Validation successful, return user info

    return None # Validation failed

Step 3: Create Login and Registration Pages

We need to provide an interface for users to register and log in. In the templates folder, create login.html and register.html files.

register.html

{% include "_header.html" %}
<form action="/users/register" method="POST" class="post-form">
  <h2>Register</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Register</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Already have an account? <a href="/auth/login">Login here</a>.
</p>
{% include "_footer.html" %}

login.html

{% include "_header.html" %}
<form action="/auth/login" method="POST" class="post-form">
  <h2>Login</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Login</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Don't have an account? <a href="/users/register">Register here</a>.
</p>
{% include "_footer.html" %}

At the same time, let's update _header.html to add links for registration and login in the top-right corner.

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      <a href="/posts/new" class="new-post-btn">New Post</a>
      <a href="/users/register" class="nav-link">Register</a>
      <a href="/auth/login" class="nav-link">Login</a>
    </nav>
  </header>
  <main>

Step 4: Implement Routing and Controller Logic

Now, we will split the routing logic into different files to make the project structure clearer.

Create a routers folder in the project's root directory.

Cut all the Post-related routes (@app.get("/posts", ...) etc.) from main.py and paste them into a routers/posts.py file.

# routers/posts.py
import uuid
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select

from database import get_session
from models import Post

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/", response_class=HTMLResponse)
def root():
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(request: Request, session: Session = Depends(get_session)):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home"})

@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post"})

@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session)
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(request: Request, post_id: uuid.UUID, session: Session = Depends(get_session)):
    post = session.get(Post, post_id)
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title})

Note: We are replacing the @app decorator with @router.

In the routers folder, create users.py to handle user registration.

# routers/users.py
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import users_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/users/register", response_class=HTMLResponse)
def show_register_form(request: Request):
    return templates.TemplateResponse("register.html", {"request": request, "title": "Register"})

@router.post("/users/register")
def register_user(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    # For simplicity, no complex validation here
    users_service.create_user({"username": username, "password": password}, session)
    return RedirectResponse(url="/auth/login", status_code=302)

In the routers folder, create auth.py to handle user login.

# routers/auth.py
from fastapi import APIRouter, Request, Depends, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import auth_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/auth/login", response_class=HTMLResponse)
def show_login_form(request: Request):
    return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})

@router.post("/auth/login")
def login(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    user = auth_service.validate_user(username, password, session)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")

    # Validation successful
    return RedirectResponse(url="/posts", status_code=302)

Finally, update main.py to remove the old routes and include the new router files.

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from database import create_db_and_tables
from routers import posts, users, auth

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating tables..")
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

# Mount the static files directory
app.mount("/static", StaticFiles(directory="public"), name="static")

# Include routers
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)

Step 5: Testing

At this point, we have completed a basic user registration and login validation logic.

Restart your project:

uvicorn main:app --reload

Visit http://localhost:3000/users/register to register.

After successful registration, you will be automatically redirected to http://localhost:3000/auth/login to log in.

You can test the results of entering correct and incorrect account credentials. For example, if you enter incorrect information, the page will show a 401 Unauthorized error.

However, the current login is just a one-time validation process; the server does not "remember" the user's login state. After closing the browser or visiting other pages, you will still be in an unauthenticated state.

In the next article, we will introduce session management to achieve true user login state persistence and restrict page access and operations based on user permissions.

Read on our blog

Related Posts: