In this article, we explore the creation of a dynamic product catalog using FastAPI, SQLAlchemy, and Pydantic.

This application serves as a practical introduction to handling CRUD operations in a RESTful API, demonstrating how to design, implement, and structure a scalable web application.

We will dissect the application into three primary files: database.py, models.py, and main.py, explaining the purpose and function of each component in the context of managing product data.

This is my introduction article on getting started with FastAPI:

Here’s link to article on FastAPI with Pydantic:

Establishing the Database Connection with database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URL = 'sqlite:///./products.db'

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False})

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

In database.py, we set up the foundation for our database interactions:

• Engine Creation: Initiates a connection to a SQLite database named products.db. SQLite is chosen for its simplicity, making it ideal for small-scale applications and demonstrations.
• SessionLocal: Defines the configuration for our database sessions, facilitating transactions and interactions with the database.
• Base: Acts as a declarative base for our model classes, serving as a registry of models and tables.

Modeling Our Data with models.py

from database import Base
from sqlalchemy import Column, Integer, String, Float, Boolean

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, primary_key=False, index=True)
    description = Column(String, index=False)
    price = Column(Float)
    is_available = Column(Boolean, default=True)

models.py outlines the structure of our product catalog in the database:

• Product Class: Defines a product model that includes an id, name, description, price, and availability status (is_available).
• Columns: Specify the attributes of our products and how they map to the database schema.

Implementing the API with main.py

from typing import Annotated
from sqlalchemy.orm import Session
from starlette import status
from pydantic import BaseModel, Field
from fastapi import FastAPI, Depends, HTTPException, Path
import models
from models import Product
from database import engine, SessionLocal

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

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

db_dependency = Annotated[Session, Depends(get_db)]

class ProductRequest(BaseModel):
    name: str = Field(min_length=1)
    description: str = Field(min_length=3, max_length=500)
    price: float = Field(gt=0)
    is_available: bool

The main.py file is where our API routes are defined, enabling operations on the product catalog:

• FastAPI Instance: Sets up our web application and routes.
• Database Connection: Utilizes get_db to provide a session for each endpoint, ensuring that database transactions are properly managed.
• Pydantic Models: Validates and structures the data for product requests and responses.

Defining Our Endpoints

@app.get('/products', status_code=status.HTTP_200_OK)
async def read_all_products(db: db_dependency):
    return db.query(Product).all()

@app.get('/product/{product_id}', status_code=status.HTTP_200_OK)
async def get_product_by_id(db: db_dependency, product_id: int = Path(gt=0)):
    ...

@app.post('/product', status_code=status.HTTP_201_CREATED)
async def create_product(db: db_dependency, product_request: ProductRequest):
    ...

@app.put('/product/{product_id}', status_code=status.HTTP_204_NO_CONTENT)
async def update_product(db: db_dependency, product_request: ProductRequest, product_id: int = Path(gt=0)):
    ...

@app.delete('/product/{product_id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(db: db_dependency, product_id: int = Path(gt=0)):
    ...

These endpoints enable the creation, retrieval, update, and deletion of products in our catalog, using db_dependency for database access and ProductRequest for data validation.

Retrieving a Product by ID

@app.get('/product/{product_id}', status_code=status.HTTP_200_OK)
async def get_product_by_id(db: db_dependency, product_id: int = Path(gt=0)):
    product = db.query(Product).filter(Product.id == product_id).first()
    if product is not None:
        return product
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Product not found')

Creating a New Product

@app.post('/product', status_code=status.HTTP_201_CREATED)
async def create_product(db: db_dependency, product_request: ProductRequest):
    new_product = Product(**product_request.model_dump())

    db.add(new_product)
    db.commit()

Updating an Existing Product

@app.put('/product/{product_id}', status_code=status.HTTP_204_NO_CONTENT)
async def update_product(db: db_dependency, product_request: ProductRequest, product_id: int = Path(gt=0)):
    product = db.query(Product).filter(Product.id == product_id).first()
    if product is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Product not found')
    
    # Update product fields from request
    for var, value in vars(product_request).items():
        setattr(product, var, value) if value else None

    db.commit()

Deleting a Product

@app.delete('/product/{product_id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(db: db_dependency, product_id: int = Path(gt=0)):
    product = db.query(Product).filter(Product.id == product_id).first()

    if product is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Product not found')
    
    db.delete(product)
    db.commit()

This completes the CRUD functionality for your product catalog application. Each endpoint performs a specific operation:

• Read all products: Lists all products from the database.
• Get a product by ID: Retrieves a single product by its ID, throwing a 404 error if the product is not found.
• Create a product: Adds a new product to the database.
• Update a product: Modifies an existing product’s details based on the provided ID. It returns a 404 error if the product does not exist.
• Delete a product: Removes a product from the database, a 404 error is returned if the product is not found.

These endpoints provide a robust API for managing a product catalog, demonstrating the power and simplicity of combining FastAPI, SQLAlchemy, and Pydantic for web development projects.

Piecing It All Together

The separation into database.py, models.py, and main.py not only organizes the application logically but also showcases best practices for building scalable and maintainable web applications.

I trust this information has been valuable to you. 🌟 Wishing you an enjoyable and enriching learning journey!