Building a RESTful API with FastAPI, SQLAlchemy & PostgreSQL

Python has various web frameworks like Flask, Django, FastAPI and more. Among them, FastAPI stands out for its simplicity, performance, and ease of use.

FastAPI is built on top of Pydantic and Starlette, which provides built-in type validation, async support. If you've worked with Flask or Django, you'll find FastAPI a bit too straightforward :)

Let's explore FastAPI and build a RESTful API with SQLAlchemy and PostgreSQL.

About `FastAPI` ⚡️

Built on Starlette (a lightweight ASGI framework for high-performance async web applications) and Pydantic (a data validation library for defining and validating data models).
API Documentation supported out-of-the-box with Swagger UI and ReDoc.
Provides async support for building high-performance applications, dependency injection support and much more.

Specific to FastAPI~

FastAPI(): Creates your application instance with automatic OpenAPI schema generation
Depends(): Dependency injection for database sessions, authentication, and more
HTTPException: HTTP error responses with status codes
Pydantic models: Automatic request validation and response serialization
response_model: Defines and validates response structure

Building Your API 🚀

We'll build a RESTful API to manage products in a store with PostgreSQL database. The API will support:

GET /products → Get all products
POST /products → Create a new product
GET /products/:id → Get a specific product
PUT /products/:id → Update product details
DELETE /products/:id → Remove a product

Let's dive in!

0. Prerequisites & Setup

Before we begin, ensure you have Python 3.8+ and Docker installed. Create a new directory for your project and set up a virtual environment:

bash

~ $ mkdir products-api && cd products-api

~/products-api $ python -m venv .venv

~/products-api $ source .venv/bin/activate  

# On Windows: .venv\Scripts\activate

Create a requirements.txt file with the necessary dependencies:

txt

annotated-doc==0.0.4
annotated-types==0.7.0
anyio==4.12.1
click==8.3.1
fastapi==0.128.0
h11==0.16.0
idna==3.11
psycopg2-binary==2.9.11
pydantic==2.12.5
pydantic-settings==2.12.0
pydantic_core==2.41.5
python-dotenv==1.2.1
SQLAlchemy==2.0.45
starlette==0.50.0
typing-inspection==0.4.2
typing_extensions==4.15.0
uvicorn==0.40.0

Install the dependencies:

bash

~/products-api $ pip install -r requirements.txt

1. Start Simple

Let's create a minimal FastAPI server to verify our setup. Create main.py:

python

from fastapi import FastAPI

app = FastAPI(
    title="Products API",
)

@app.get("/")
def root():
    return {
        "message": "Welcome to Products API", "status": "active"
        }

Run the server:

bash

~/products-api $ uvicorn main:app --reload

Visit http://localhost:8000 to see your API response, and http://localhost:8000/docs for the automatic interactive documentation.

2. Database Configuration

💡 We'll use PostgreSQL with Docker for easy setup. SQLAlchemy ORM will handle database operations.

Create docker-compose.yml for PostgreSQL:

yaml

services:
  db:
    container_name: products_db
    image: postgres:15.4-alpine
    environment:
      - POSTGRES_USER=USER_NAME
      - POSTGRES_PASSWORD=PASSWORD
      - POSTGRES_DB=DB_NAME
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    networks:
      - app_network
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

Create a .env file for database configuration:

bash

DATABASE_URL=postgresql://USERNAME:PASSWORD@localhost:5432/DB_NAME

💡 Replace USERNAME, PASSWORD, and DB_NAME with your actual database credentials.

Start the database:

bash

~/products-api $ docker-compose up -d

3. Setting Up SQLAlchemy

Create settings.py to manage environment variables using Pydantic:

python

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_url: str

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

Create database.py for database connection:

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from settings import settings

engine = create_engine(settings.database_url)
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

4. Define Database Models

Create database_models.py with SQLAlchemy models:

python

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, Float

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[str] = mapped_column(String(255))
    price: Mapped[float] = mapped_column(Float)
    quantity: Mapped[int] = mapped_column(Integer)

5. Define Pydantic Models

Create models.py for request/response validation:

python

from pydantic import BaseModel, Field
from typing import Optional

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str = Field(..., min_length=1, max_length=255)
    price: float = Field(..., gt=0, description="Must be positive")
    quantity: int = Field(..., ge=0, description="Must be non-negative")

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, min_length=1, max_length=255)
    price: Optional[float] = Field(None, gt=0)
    quantity: Optional[int] = Field(None, ge=0)

class ProductResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    quantity: int

    class Config:
        from_attributes = True  # Enables SQLAlchemy model conversion

Why separate models?

ProductCreate: Client doesn't provide id (database generates it)
ProductUpdate: All fields optional for partial updates
ProductResponse: Includes id from database

6. Building the API Endpoints

Now let's build the complete API in main.py:

python

import logging
from fastapi import FastAPI, Depends, HTTPException, status
from models import ProductCreate, ProductUpdate, ProductResponse
import database_models
from database import session, engine
from sqlalchemy.orm import Session

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Products API",
    description="A simple FastAPI backend for managing products",
    version="1.0.0"
)

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

# Dependency: Database session management
def get_db():
    db = session()
    try:
        yield db
    finally:
        db.close()

💡 In real-world projects, consider using Alembic for database migrations.

6.1. GET /products - Retrieve All Products

python

@app.get("/products", response_model=list[ProductResponse], tags=["Products"])
def get_products(db: Session = Depends(get_db)):
    try:
        logger.info("Fetching all products")

        products = db.query(database_models.Product).all()
        return products
    except Exception as e:
        logger.error(f"Error fetching products: {e}")
        
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to retrieve products"
        )

Key concepts:

response_model=list[ProductResponse]: Validates response structure
Depends(get_db): Injects database session automatically
HTTPException: Returns proper HTTP error responses

6.2. GET /products/:id - Get Single Product

python

@app.get("/products/{product_id}", response_model=ProductResponse, tags=["Products"])
def get_product(product_id: int, db: Session = Depends(get_db)):
    logger.info(f"Fetching product with ID: {product_id}")

    product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    return product

6.3. POST /products - Create New Product

python

@app.post(
    "/products",
    response_model=ProductResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Products"]
)
def add_product(product: ProductCreate, db: Session = Depends(get_db)):
    try:
        logger.info(f"Creating new product: {product.name}")

        # Pydantic validates the request automatically
        db_product = database_models.Product(**product.model_dump())
        db.add(db_product)
        db.commit()
        db.refresh(db_product)  # Get the generated ID

        logger.info(f"Product created with ID: {db_product.id}")
        
        return db_product
    
    except Exception as e:
        logger.error(f"Error creating product: {e}")
    
        db.rollback()
    
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to create product"
        )

What's happening here?

FastAPI automatically validates the request body against ProductCreate
Invalid data (negative price, empty name) returns 422 error
db.refresh() fetches the auto-generated ID from database

6.4. PUT /products/:id - Update Product

python

@app.put("/products/{product_id}", response_model=ProductResponse, tags=["Products"])
def update_product(product_id: int, product: ProductUpdate, 
db: Session = Depends(get_db)):
    logger.info(f"Updating product with ID: {product_id}")

    db_product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not db_product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    try:
        # Update only fields that were provided
        update_data = product.model_dump(exclude_unset=True)
       
        for field, value in update_data.items():
            setattr(db_product, field, value)

        db.commit()
        db.refresh(db_product)
    
        return db_product
    
    except Exception as e:
        logger.error(f"Error updating product: {e}")
       
        db.rollback()
       
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to update product"
        )

💡 exclude_unset=True only includes fields that were actually provided in the request, enabling partial updates.

6.5. DELETE /products/:id - Remove Product

python

@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Products"])
def delete_product(product_id: int, db: Session = Depends(get_db)):
    logger.info(f"Deleting product with ID: {product_id}")

    product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    try:
        db.delete(product)
        db.commit()
    
        logger.info(f"Product deleted successfully")
    
        return None  # 204 No Content
    
    except Exception as e:
        logger.error(f"Error deleting product: {e}")
    
        db.rollback()
    
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to delete product"
        )

7. Makefile for Convenience 🛠

Create a Makefile to simplify common tasks:

makefile

.PHONY: help install run db-up db-down clean setup env

help:
	@echo "Available commands:"
	@echo "  make install   - Install dependencies"
	@echo "  make run       - Run the API server"
	@echo "  make db-up     - Start PostgreSQL"
	@echo "  make db-down   - Stop PostgreSQL"
	@echo "  make setup     - Complete setup"

install:
	pip install -r requirements.txt

run:
	uvicorn main:app --reload

db-up:
	docker-compose up -d
	@echo "Database is ready!"

db-down:
	docker-compose down

setup:
	@make install
	@make db-up
	@echo "Setup complete! Run 'make run' to start."

env:
	@if [ ! -f .env ]; then \
		echo "DATABASE_URL=postgresql://admin:admin123@localhost:5432/products_db" > .env; \
		echo ".env file created"; \
	fi

Now you can use simple commands:

bash

~/products-api $ make setup  # Setup project
~/products-api $ make run    # Start your API

8. API Documentation 🔨

Navigate to http://localhost:8000/docs or http://localhost:8000/redoc to explore the automatically generated API documentation.

Try these:

Create a product:

json

POST /products
{
  "name": "Laptop",
  "description": "MacBook Pro",
  "price": 1999.99,
  "quantity": 10
}

Get all products:

GET /products

Update a product:

json

PUT /products/1
{
  "price": 1799.99
}

Delete a product:

DELETE /products/1

🔗 Check out the full code on GitHub

TL;DR

FastAPI provides automatic validation, built-in support for documentation, and type safety out of the box
Use separate Pydantic models for create, update, and response operations
SQLAlchemy ORM abstracts database operations
Depends() enables dependency injection for database sessions
Learn more about FastAPI from the official documentation

About FastAPI ⚡️

Specific to FastAPI~

Building Your API 🚀

0. Prerequisites & Setup

1. Start Simple

2. Database Configuration

3. Setting Up SQLAlchemy

4. Define Database Models

5. Define Pydantic Models

6. Building the API Endpoints

6.1. GET /products - Retrieve All Products

6.2. GET /products/:id - Get Single Product

6.3. POST /products - Create New Product

6.4. PUT /products/:id - Update Product

6.5. DELETE /products/:id - Remove Product

7. Makefile for Convenience 🛠

8. API Documentation 🔨

TL;DR

About `FastAPI` ⚡️