API: Beginner to Advance: Building User Registration in FastAPI- Part 8

Let’s learn how can we allow users to create accounts, log in, and associate posts with their specific accounts.

17 min readAug 24, 2024

This is a continuation blog of my API: Beginner to Advance. I strongly recommend reading my previous blog before continuing to read to strengthen your concepts on PostgreSQL and building basic APIs.

Talib

APIs: Beginner to Advance

View list

9 stories

Find the full code: https://github.com/talibilat/APIs

Define a User Model in SQLAlchemy
Create a User Schema for Validation
Create a User in the Database
Secure the Password
Define a User Response Model
Refactor Password Hashing Logic into a Utility Function
Organising Routes with Routers and Prefixes
Authentication Overview
JWT Authentication Flow
JWT Token Structure
Why the Signature Matters
Token Tampering Prevention
Password Hashing
Code JWT Authentication
JWT Authentication and Securing Routes

Define a User Model in SQLAlchemy

To handle user registration, we need a users table in our PostgreSQL database. Since we are using SQLAlchemy as our ORM, we will define a new SQLAlchemy model for users, similar to what we did for posts.

Create the User Model:
Start by defining a User class that will extend SQLAlchemy's Base model. This class will define the users table schema.

# 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)
    email = Column(String, unique=True, nullable=False)
    password = Column(String, nullable=False)
    created_at = Column(TIMESTAMP(timezone=True), server_default=text('now()'), nullable=False)

email: This field will store the user’s email. We set it to unique=True to ensure that each email address can only be used once.
password: This field will store the user’s password.
created_at: A timestamp that records when the user account was created.

Test the Model in PostgreSQL:

After defining the model, we need to ensure that the table is created properly. FastAPI will create the table automatically when the app starts:

# In Postgres 
SELECT * FROM users;

You can manually add some test data in PostgreSQL to verify that the table is working as expected.

Create a User Schema for Validation

Next, we need to define a schema for validating the data users send when creating an account. This schema will ensure that users provide valid data (like a valid email address) when registering.

Define a UserCreate Schema:
Create a Pydantic model that represents the structure of the request for creating a user.

# schemas.py
from pydantic import BaseModel, EmailStr

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

EmailStr: This is a Pydantic field that ensures the provided value is a valid email address.

Create a User in the Database

We now need to implement a path operation to allow users to create accounts. This function will handle receiving user data, validating it, and saving it to the database.

Define a create_user Function:
In the main.py file, define the path operation for creating a user:

# main.py
@app.post("/users/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Hash the password here (discussed later)
    new_user = models.User(**user.dict())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

user.dict(): Converts the Pydantic model into a dictionary, which we can pass to SQLAlchemy.
db.add(new_user): Adds the new user to the database.
db.commit(): Commits the transaction to save the new user.
db.refresh(new_user): Reloads the user from the database to include any auto-generated fields (like id and created_at).

Test User Creation

Use a tool like Postman to test the user creation endpoint. Send a POST request to /users/ with the following JSON body:

{
    "email": "talib@gmail.com",
    "password": "talib@123"
}

The expected response should contain the user’s id, email, and created_at timestamp. Ensure that the password is not returned.

Secure the Password

Storing passwords in plain text is highly insecure. Instead, we will hash the password before saving it to the database. We can use a library like bcrypt to hash passwords.

Install bcrypt:
Install the bcrypt library for password hashing:

pip install bcrypt

Hash the Password:

Update the create_user function to hash the password before storing it:

from passlib.context import CryptContext

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

def hash_password(password: str):
    return pwd_context.hash(password)

@app.post("/users/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    hashed_password = hash_password(user.password)
    new_user = models.User(email=user.email, password=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

pwd_context.hash(password): This will hash the provided password before saving it to the database.

Define a User Response Model

When returning user data, we should not include sensitive information like the password. To handle this, we can define a Pydantic model for shaping the response.

Create a UserOut Schema:

Define a response model that excludes the password:

# schemas.py
class UserOut(BaseModel):
    id: int
    email: EmailStr
    created_at: datetime

    class Config:
        orm_mode = True

orm_mode = True: Tells Pydantic to read data even if it is returned from an ORM model (like SQLAlchemy).

Apply the Response Model:

Update the create_user path operation to use this response model:

@app.post("/users/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    hashed_password = hash_password(user.password)
    new_user = models.User(email=user.email, password=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

By following these steps, you have successfully implemented user registration in your FastAPI application. You now have a PostgreSQL users table, a secure way to store user data, and a well-structured API that returns only necessary information to the client.

Refactor Password Hashing Logic into a Utility Function

To improve the maintainability of our code, we extracted the password hashing logic into a utility function in a separate utils.py file. This allows us to easily reuse the hashing function across different parts of the application.

Create utils.py:

In the utils.py file, create a function for password hashing:

# utils.py
from passlib.context import CryptContext

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

def hash_password(password: str):
    return pwd_context.hash(password)

pwd_context.hash(password): This hashes the provided password using the bcrypt algorithm.

Update main.py to Use the Utility Function:

In main.py, import the utils module and use the hash_password function when creating a user:

# main.py
from . import utils

@app.post("/users/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    hashed_password = utils.hash_password(user.password)
    new_user = models.User(email=user.email, password=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

This keeps the password hashing logic separate and organized in one place (utils.py), improving code readability and maintainability.

Create a Route to Fetch User Information by ID

Now, we will add a route to allow users (or the app) to fetch information about a specific user by their ID. This is useful for various reasons, such as checking user profile information or verifying authentication.

Define the get_user Function:

Create a path operation to fetch user details based on their ID:

# main.py
from . import utils

@app.post("/users/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    hashed_password = utils.hash_password(user.password)
    new_user = models.User(email=user.email, password=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

Find the full code: https://github.com/talibilat/APIs

Organising Routes with Routers and Prefixes

In this section, we will continue focusing on improving the structure of the FastAPI project by utilizing routers and prefixes. This will help make our code cleaner, easier to maintain, and reduce duplication of code.

Step 1: Organize Routes into Separate Files

As the project grows, keeping all route definitions in the main.py file can make the project messy and harder to manage. We’ll move route definitions related to posts and users into separate files under a routers/ directory.

Create a routers/ Directory:

Create a new directory called routers/ to hold separate route files for posts and users.

mkdir routers

Create post.py and user.py:

Inside the routers/ directory, create two Python files: post.py and user.py.

touch routers/post.py routers/user.py

Move Post Routes to post.py: Open main.py, cut the post-related routes, and paste them into routers/post.py.

Move User Routes to user.py: Similarly, move the user-related routes from main.py to routers/user.py.

Step 2: Fix Imports and Define Routers

Fix Imports in post.py and user.py: Since the new files are inside the routers/ directory, we need to adjust the import paths for models, schemas, and database.

In routers/post.py and routers/user.py, we import necessary modules like this:

# post.py or user.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, schemas
from ..database import get_db

The .. indicates moving up one directory (from routers/ to the main app/ directory).

Define API Router Objects:

Replace the app object with APIRouter in both post.py and user.py.

# post.py
router = APIRouter()

@router.get("/")
def get_posts(db: Session = Depends(get_db)):
    # your logic here

Similarly, in user.py, do the same:

# user.py
router = APIRouter()

@router.post("/", response_model=schemas.UserOut, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # your logic here

Step 3: Include Routers in `main.py`

Include Routers in main.py: In the main.py file, import the router objects from post.py and user.py and include them using app.include_router().

# main.py
from fastapi import FastAPI
from .routers import post, user

app = FastAPI()

# Include the routers
app.include_router(post.router)
app.include_router(user.router)

This tells FastAPI to include all routes defined in the post and user files.

Step 4: Use Route Prefixes to Simplify Paths

To avoid repetition in defining paths for the routes (e.g., /posts/ or /users/), we can use the prefix argument when defining the router.

Add Prefix to post.py: Add a prefix to the API Router in post.py:

# post.py
router = APIRouter(prefix="/posts")

@router.get("/")
def get_posts():
    return {"message": "Get all posts"}

Now all the routes in this file will automatically be prefixed with /posts, eliminating the need to specify it manually in every route definition.

Add Prefix to user.py: Similarly, add a prefix to user.py

# user.py
router = APIRouter(prefix="/users")

@router.post("/", response_model=schemas.UserOut)
def create_user():
    return {"message": "User created"}

This ensures that all routes in user.py will be prefixed with /users.

Update Route Definitions:

After adding prefixes, the route definitions can be simplified. For instance, instead of:

@router.get("/posts/{id}")

We can now simply write:

@router.get("/{id}")

Since the router already knows that all routes start with /posts, we only need to define the additional dynamic part (e.g., /{id}).

Step 5: Test Routes

After making the changes, restart the FastAPI server and test the endpoints to ensure everything works.

Test Post Endpoints:

GET /posts/: Retrieve all posts.
GET /posts/{id}: Retrieve a specific post by ID.
POST /posts/: Create a new post.

Test User Endpoints:

POST /users/: Create a new user.
GET /users/{id}: Retrieve a user by ID.

All routes should work as expected, and your code should now be much cleaner and easier to maintain.

Authentication Overview

Authentication is a key concept in developing APIs or any application, and there are two main ways to handle it: session-based authentication and JWT-based authentication. These methods have distinct characteristics:

Session-based authentication:

In this method, the server stores information (session data) either in the database or in memory. This session data helps track whether a user is logged in or logged out. When a user logs in, the server creates a session and stores it, and when the user logs out, the session is cleared.
The session is often stored as a session ID in a cookie on the client side, and the server maintains state by keeping track of these sessions.

JWT-based authentication:

This method is stateless, meaning there’s nothing stored on the backend, API, or database to track whether the user is logged in or out.
The JWT (JSON Web Token) is stored on the client side and passed to the API during each request.
The power of JWT lies in the token itself, which keeps track of the user’s login state without needing backend storage.
The token is generated by the server and sent to the client upon successful login. The client stores the token (typically in localStorage or sessionStorage) and sends it in the Authorization header of each request for accessing protected routes.

Find the full code: https://github.com/talibilat/APIs

JWT Authentication Flow

Let’s break down the JWT authentication process:

Login Request:

The client sends a POST request to the /login endpoint, passing in the user’s credentials (usually email and password).
These credentials can vary (e.g., username/password or email/password), but in this example, we’ll use email and password.

Credential Verification:

The server checks if the credentials provided by the user match those in the database.
If the email and password match, the server proceeds to the next step.

JWT Token Creation:

Once credentials are verified, the server creates a JWT token for the user. This token contains important information and is sent back to the client as part of the login response.
This token is just a string from the client’s perspective, and the client doesn’t need to know the contents of the token. The token itself is handled by the API.

Accessing Protected Resources:

Once the client has the token, it can be used to access protected resources (like the /posts endpoint).
For each request to a protected route, the client sends the token in the Authorization header of the request.
The API verifies the token’s validity and grants access to the resource if the token is valid.

JWT Token Structure

A JWT token consists of three main parts:

Header:

The header contains metadata about the token, including:

The algorithm used to sign the token (commonly HS256, HMAC using SHA-256).
The type of token (JWT).

Payload:

The payload contains claims, which are pieces of information that are embedded within the token. These claims can be:

Registered claims: Standard claims like issuer (iss), subject (sub), expiration time (exp), etc.
Public claims: Custom claims, like the user’s ID and role (e.g., admin, regular user).
The payload can include any piece of information you want, but it’s important to avoid putting sensitive information (like passwords or secrets) because the token is not encrypted.
Common claims include:
User ID: Allows the API to know which user made the request.
User role: Identifies if the user is an admin or a regular user.

Signature:

The signature is critical for ensuring the token’s integrity and is generated by:

Combining the header and payload with a secret key (stored only on the server) using the algorithm specified in the header (e.g., HS256).
The secret key is known only to the server and is used to ensure the token hasn’t been tampered with.
If someone tries to alter the token (like changing the user’s role), the signature will no longer match, and the token will be rejected by the API.

Verifying the Token:

When the client makes a request to a protected resource, the token is included in the Authorization header.
The server takes the token, extracts the header and payload, and uses the secret key to regenerate the signature.
The regenerated signature is then compared with the one in the token. If they match, the token is valid, and access is granted. If they don’t match, the token is rejected.

Why the Signature Matters

The signature is crucial because it ensures the data integrity of the token. Here’s how:

The token contains three parts: header, payload, and signature.
If someone tries to alter the token’s payload (for example, changing the user role to “admin”), they cannot generate a valid signature because they don’t have access to the secret key.
The server compares the signature sent with the token against the signature it generates using the secret key. If the signatures don’t match, the token is invalid.

Token Tampering Prevention

If a malicious user tries to tamper with the token, such as changing the payload (e.g., changing their role from “user” to “admin”):

They would need to create a new signature to match the modified token.
However, they don’t have the secret key, so they can’t generate a valid signature.
The server will detect this tampering by comparing the signatures and reject the token.

Example:

Imagine a token with the payload containing the user ID and role:

{
  "user_id": 123,
  "role": "user"
}

If the user tries to change the role to “admin”, the original signature, which was created using the secret key, won’t match the altered token’s signature. As a result, the server will reject the modified token.

Password Hashing

When a user attempts to log in, they send their plain text password to the server. However, passwords in the database are stored as hashed passwords for security reasons. The process for checking passwords is as follows:

The server retrieves the hashed password from the database.
The plain text password provided by the user is hashed.
The server compares the hashed version of the provided password with the hashed password in the database.
If the hashed passwords match, the credentials are correct, and the user is authenticated. If not, the login fails.

This approach ensures that even if someone gains access to the database, they won’t see the plain text passwords, as they are stored in a hashed form, which is one-way and cannot be reversed.

Final Steps in Token-based Authentication:

Once the credentials are validated, the server creates a JWT token.
The client stores the token and sends it in the header for future requests to protected routes.
The API verifies the token’s integrity by comparing signatures and grants access if valid.

This process ensures a stateless and secure method of authentication, where the client is responsible for storing and sending the token, while the server validates it on each request.

Code JWT Authentication

To create the login path operation, we’re going to store it in a separate auth.py file rather than combining it with user routes. This allows for better organization by separating authentication logic from user management.

Verify the password:

Implement a password verification function in a utility module (utils.py) to compare the provided plain text password with the hashed password stored in the database.

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

Schema for User Login:

Create a schema in schemas.py for login credentials, specifying the fields email and password. Use EmailStr for email validation.

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

Create a new router for authentication:

In auth.py, start by importing the necessary components from FastAPI: APIRouter, Depends, HTTPException, and status.

Define the login path operation: The login operation will be a POST request, as it involves sending the user’s credentials (email and password) for authentication.

from fastapi import APIRouter, Depends, status, HTTPException, Response
from sqlalchemy.orm import Session
from .. import database, schemas, models, utils

router = APIRouter(tags=['Authentication'])


@router.post('/login')
def login(user_cerd: schemas.UserLogin, db: Session = Depends(database.get_db)):
    
    user = db.query(models.User).filter(models.User.email == user_cerd.email).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Invalid Credentials")
    if not utils.verify(user_cerd.password,user.password):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Invalid Credentials")

Set up database dependencies: Import the necessary database dependencies, including Session from SQLAlchemy and the get_db function from the database module to access the database in the login function.

Query the user from the database: Use the provided email to query the users table and retrieve the user data

Handle invalid credentials: If no user is found with the provided email, raise an HTTPException with a 404 status code and a message like "Invalid credentials."

Verify the password: Use the verify function inside the login path to validate the password:

Handle successful login: Once the credentials are verified, you will generate a JWT token (not implemented yet) and return it to the user.

For now, return a placeholder token to ensure everything works:

return "Success"

Register the router in `main.py`:

Import the auth.py router in main.py and add it to the FastAPI application:

app.include_router(auth.router)

Testing the API:

After setting up the endpoint and registering it in main.py, test it by:
Sending a POST request to /login with a JSON body containing the user's email and password.
Verify that the correct token is returned for valid credentials and an “Invalid credentials” message is returned for incorrect email/password combinations.

By following this process, you separate concerns by creating a dedicated auth.py for authentication, implement password verification using bcrypt, and prepare the code for generating JWT tokens in the next steps.

Find the full code: https://github.com/talibilat/APIs

JWT Authentication and Securing Routes

This section will secure certain routes and ensure that only authenticated users can access them. We’ll also cover using JWT tokens to validate requests and verify users.

Step 1: JWT Token Creation and Verification

We’ve already set up the JWT token creation using the OAuth2PasswordBearer mechanism and built functions for creating access tokens and verifying them. These tokens are sent by users to prove their authentication when accessing certain endpoints.

The create_access_token function takes in user data and generates a JWT with an expiration time, while verify_access_token decodes and verifies the token provided by users when they access protected routes.

Step 2: Securing the Routes

Now, we need to secure our routes by requiring valid tokens for certain actions, like creating, updating, or deleting posts, and retrieving users.

Update post.py for Protected Routes:

In the routers/post.py, we will modify the routes such that users must be authenticated to create, update, and delete posts.

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, schemas, oauth2
from ..database import get_db

router = APIRouter(prefix="/posts")

@router.post("/", response_model=schemas.Post)
def create_post(post: schemas.PostCreate, db: Session = Depends(get_db), 
                current_user: models.User = Depends(oauth2.get_current_user)):
    # Now we have access to the current user
    new_post = models.Post(owner_id=current_user.id, **post.dict())
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post

Here, we have added a dependency on the get_current_user function from oauth2.py. This function will verify the token and return the current authenticated user.

Require Authentication for Other Routes: We’ll add the same authentication requirement for updating and deleting posts:

@router.put("/{id}", response_model=schemas.Post)
def update_post(id: int, post: schemas.PostCreate, db: Session = Depends(get_db), 
                current_user: models.User = Depends(oauth2.get_current_user)):
    post_query = db.query(models.Post).filter(models.Post.id == id, models.Post.owner_id == current_user.id)
    post_to_update = post_query.first()
    if not post_to_update:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
    post_query.update(post.dict(), synchronize_session=False)
    db.commit()
    return post_to_update

@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int, db: Session = Depends(get_db), 
                current_user: models.User = Depends(oauth2.get_current_user)):
    post_query = db.query(models.Post).filter(models.Post.id == id, models.Post.owner_id == current_user.id)
    post_to_delete = post_query.first()
    if not post_to_delete:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
    db.delete(post_to_delete)
    db.commit()

The delete and update routes will also require the user to be authenticated and match the owner_id before making changes.

Step 3: Fetching the User’s Information

In the get_current_user function (found in oauth2.py), we fetch the user’s information using their ID encoded in the token. This step is critical in verifying whether the user is valid and authorized to perform certain actions.

We fetch the user from the database using their id, which is extracted from the token’s payload.

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    token_data = verify_access_token(token, credentials_exception)
    user = db.query(models.User).filter(models.User.id == token_data.id).first()
    if user is None:
        raise credentials_exception
    return user

Testing Secured Routes

Now that we’ve secured the routes, let’s test them:

Get an Access Token: First, you need to log in by providing valid credentials to the login route to get a JWT token. This token is valid for a set amount of time (e.g., 30 minutes).

POST /login
{
    "username": "user@example.com",
    "password": "password"
}

Use the Token in Protected Routes: To access a protected route (e.g., creating a post), include the token in the Authorization header of your request:

Authorization: Bearer your.jwt.token.here

If the token is valid and not expired, the user can create a post. If not, they will receive a 401 Unauthorized error.

By securing routes with JWT tokens, we’ve ensured that only authenticated users can perform sensitive operations. The use of get_current_user as a dependency allows us to easily check if a user is authenticated before allowing them to proceed with certain actions like creating or deleting posts.

With this setup, we can build more complex APIs that ensure secure access to resources. In the next steps, we can add refresh tokens, more complex authorization mechanisms, and user roles for more control over the application’s flow.

Thank you for reading!