14.3 Building a CRUD API with SQLAlchemy

NicePython

Oct 4, 2024 — 5 min read

In this section, we will learn how to build a simple CRUD (Create, Read, Update, Delete) API using SQLAlchemy to interact with a PostgreSQL database. We'll use FastAPI, a modern web framework for building APIs in Python, alongside SQLAlchemy for managing the database operations.

The goal is to create an API that allows clients to:

Create new records in the database.
Read existing records from the database.
Update existing records.
Delete records.

14.3.1 Setting Up the Environment

First, you'll need to install the required libraries:

FastAPI: A Python web framework for building APIs.
SQLAlchemy: For ORM database management.
Uvicorn: An ASGI server to run FastAPI.
Pydantic: A library used by FastAPI to validate data.

Install Dependencies:

pip install fastapi uvicorn sqlalchemy psycopg2

These dependencies include:

FastAPI: Web framework for building the CRUD API.
Uvicorn: A fast ASGI server for serving the FastAPI application.
SQLAlchemy: Database ORM for managing models and queries.
psycopg2: PostgreSQL adapter for Python, used by SQLAlchemy.

14.3.2 Setting Up the Project Structure

Let's create a simple folder structure for our project:

crud_api_project/
│
├── app/
│   ├── __init__.py
│   ├── database.py
│   ├── models.py
│   ├── crud.py
│   └── main.py
│
└── requirements.txt

14.3.3 Connecting to the PostgreSQL Database

First, configure the database connection in the database.py file using SQLAlchemy.

`app/database.py`:

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

# PostgreSQL database connection URL
SQLALCHEMY_DATABASE_URL = "postgresql://testuser:password@localhost/testdb"

# Create the SQLAlchemy engine
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# SessionLocal class to create a new session for each request
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for creating models
Base = declarative_base()

# Dependency for creating a new session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

In this file:

create_engine() connects to the PostgreSQL database.
SessionLocal() creates a new session for each API request.
get_db() is a dependency function to manage the session lifecycle in API requests.

14.3.4 Defining the SQLAlchemy Models

Now, define the models representing the database schema in models.py. We'll create a simple User model with columns like id, name, and email.

`app/models.py`:

from sqlalchemy import Column, Integer, String
from .database import Base

# Define the User model
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

In this file:

The User class defines the database schema for the users table.
The table contains columns for id, name, and email.

14.3.5 Creating the CRUD Operations

Next, we create functions to handle CRUD operations using SQLAlchemy. This code will be written in crud.py and will be used by the API routes to interact with the database.

`app/crud.py`:

from sqlalchemy.orm import Session
from . import models

# Create a new user
def create_user(db: Session, name: str, email: str):
    db_user = models.User(name=name, email=email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# Get a user by ID
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

# Get all users
def get_users(db: Session):
    return db.query(models.User).all()

# Update a user's data
def update_user(db: Session, user_id: int, name: str, email: str):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user:
        user.name = name
        user.email = email
        db.commit()
        db.refresh(user)
    return user

# Delete a user by ID
def delete_user(db: Session, user_id: int):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

This file defines the core CRUD operations:

Create a new user.
Read a specific user or list all users.
Update an existing user.
Delete a user.

14.3.6 Creating Pydantic Schemas for Validation

In FastAPI, Pydantic models are used to define the structure and validate the input data for API requests. We will create two Pydantic models: one for the input when creating or updating a user and another for the output when returning user data.

`app/schemas.py`:

from pydantic import BaseModel

# User schema for input validation
class UserCreate(BaseModel):
    name: str
    email: str

# User schema for output
class User(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True

In this file:

UserCreate is used for validating user creation input.
User is used for serializing the output data (ensuring compatibility with SQLAlchemy ORM models).

14.3.7 Creating the API Endpoints

Now, we'll create API endpoints for each CRUD operation in main.py using FastAPI.

`app/main.py`:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import engine, get_db

# Create the FastAPI app
app = FastAPI()

# Create database tables on startup
models.Base.metadata.create_all(bind=engine)

# Create a user
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.create_user(db=db, name=user.name, email=user.email)
    return db_user

# Get a user by ID
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# Get all users
@app.get("/users/", response_model=list[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return crud.get_users(db=db)

# Update a user
@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db=db, user_id=user_id, name=user.name, email=user.email)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# Delete a user
@app.delete("/users/{user_id}", response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.delete_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

In this file:

/users/: POST route to create a new user.
/users/{user_id}: GET route to retrieve a user by ID.
/users/: GET route to retrieve all users.
/users/{user_id}: PUT route to update a user by ID.
/users/{user_id}: DELETE route to delete a user by ID.

14.3.8 Running the API

To run the FastAPI app, use Uvicorn as the ASGI server.

uvicorn app.main:app --reload

You can now interact with the API by visiting http://localhost:8000 and using tools like cURL, Postman, or the interactive Swagger UI provided by FastAPI at http://localhost:8000/docs.

**14.3.9 Testing the API

You can test the API using the following cURL commands or Postman requests.

1. Create a User (POST):

curl -X POST "http://localhost:8000/users/" -H "Content-Type: application/json" -d '{"name": "John", "email": "john@example.com"}'

2. Get a User (GET):

curl "http://localhost:8000/users/1"

3. Get All Users (GET):

curl "http://localhost:8000/users/"

4. Update a User (PUT):

curl -X PUT "http://localhost:8000/users/1" -H "Content-Type: application/json" -d '{"name": "John Doe", "email": "johndoe@example.com"}'

5. Delete a User (DELETE):

curl -X DELETE "http://localhost:8000/users/1"

14.3.10 Summary

In this section, we built a CRUD API using FastAPI and SQLAlchemy with PostgreSQL as the database. We covered:

Creating models using SQLAlchemy.
Writing CRUD operations to interact with the database.
Using Pydantic for request validation and response serialization.
Creating API endpoints for creating, reading, updating, and deleting data.

This simple CRUD API demonstrates how to efficiently manage database records using SQLAlchemy in combination with FastAPI.

14.3 Building a CRUD API with SQLAlchemy

NicePython

14.3.1 Setting Up the Environment

Install Dependencies:

14.3.2 Setting Up the Project Structure

14.3.3 Connecting to the PostgreSQL Database

`app/database.py`:

14.3.4 Defining the SQLAlchemy Models

`app/models.py`:

14.3.5 Creating the CRUD Operations

`app/crud.py`:

14.3.6 Creating Pydantic Schemas for Validation

`app/schemas.py`:

14.3.7 Creating the API Endpoints

`app/main.py`:

14.3.8 Running the API

**14.3.9 Testing the API

1. Create a User (POST):

2. Get a User (GET):

3. Get All Users (GET):

4. Update a User (PUT):

5. Delete a User (DELETE):

14.3.10 Summary

Read more

15.1 Unit Testing with unittest in Python

Python 4 Release Date

What's new in Python 3.15

14.5 SQLAlchemy Complex Queries

14.3.1 Setting Up the Environment

Install Dependencies:

14.3.2 Setting Up the Project Structure

14.3.3 Connecting to the PostgreSQL Database

app/database.py:

14.3.4 Defining the SQLAlchemy Models

app/models.py:

14.3.5 Creating the CRUD Operations

app/crud.py:

14.3.6 Creating Pydantic Schemas for Validation

app/schemas.py:

14.3.7 Creating the API Endpoints

app/main.py:

14.3.8 Running the API

**14.3.9 Testing the API

1. Create a User (POST):

2. Get a User (GET):

3. Get All Users (GET):

4. Update a User (PUT):

5. Delete a User (DELETE):

14.3.10 Summary

Read more

15.1 Unit Testing with unittest in Python

Python 4 Release Date

What's new in Python 3.15

14.5 SQLAlchemy Complex Queries

`app/database.py`:

`app/models.py`:

`app/crud.py`:

`app/schemas.py`:

`app/main.py`: