SQLAlchemy Sync vs Async - FastAPI Demo

A comprehensive FastAPI application demonstrating CRUD operations using both synchronous and asynchronous SQLAlchemy with PostgreSQL. Perfect for learning the differences between the two approaches!

🎯 Project Overview

This project showcases:

• ✅ Complete CRUD operations (Create, Read, Update, Delete)
• 🔄 Side-by-side comparison of sync vs async implementations
• 🐘 PostgreSQL database with Docker Compose
• 🚀 FastAPI with automatic API documentation
• 📦 Clean project structure following best practices

📁 Project Structure

sqlalchemy-async-vs-sync/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI application entry point
│   ├── config.py               # Configuration settings
│   ├── models.py               # SQLAlchemy models
│   ├── schemas.py              # Pydantic schemas
│   ├── database_sync.py        # Synchronous database setup
│   ├── database_async.py       # Asynchronous database setup
│   └── routers/
│       ├── __init__.py
│       ├── sync_router.py      # Synchronous CRUD endpoints
│       └── async_router.py     # Asynchronous CRUD endpoints
├── docker-compose.yml          # PostgreSQL container setup
├── requirements.txt            # Python dependencies
├── .env.example               # Environment variables template
├── .gitignore                 # Git ignore rules
└── README.md                  # This file

🚀 Quick Start

Prerequisites

• Python 3.8+
• Docker and Docker Compose
• Git
• uv (Python package installer - faster than pip)

Installation

1. Clone the repository

   git clone <your-repo-url>
   cd sqlalchemy-async-vs-sync

2. Install uv (if not already installed)

   # Windows PowerShell
   pip install uv

3. Create virtual environment and install dependencies

   uv venv
   .venv\Scripts\Activate.ps1
   uv sync

4. Set up environment variables

   Copy-Item .env.example .env

5. Start PostgreSQL with Docker Compose

   docker-compose up -d

6. Run the application

   uvicorn app.main:app --reload

The API will be available at http://localhost:8000

📚 API Documentation

Once the application is running, visit:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

🔌 API Endpoints

Synchronous Endpoints (using sync SQLAlchemy)

Method	Endpoint	Description
POST	/sync/items	Create a new item
GET	/sync/items	Get all items (with pagination)
GET	/sync/items/{item_id}	Get a specific item
PUT	/sync/items/{item_id}	Update an item
DELETE	/sync/items/{item_id}	Delete an item

Asynchronous Endpoints (using async SQLAlchemy)

Method	Endpoint	Description
POST	/async/items	Create a new item
GET	/async/items	Get all items (with pagination)
GET	/async/items/{item_id}	Get a specific item
PUT	/async/items/{item_id}	Update an item
DELETE	/async/items/{item_id}	Delete an item

📝 Example Usage

Create an Item (using curl)

Synchronous:

curl -X POST "http://localhost:8000/sync/items" \
  -H "Content-Type: application/json" \
  -d '{"name": "Laptop", "description": "Gaming laptop", "price": 1500}'

Asynchronous:

curl -X POST "http://localhost:8000/async/items" \
  -H "Content-Type: application/json" \
  -d '{"name": "Laptop", "description": "Gaming laptop", "price": 1500}'

Get All Items (using curl)

Synchronous:

curl "http://localhost:8000/sync/items"

Asynchronous:

curl "http://localhost:8000/async/items"

Get Single Item (using curl)

curl "http://localhost:8000/sync/items/1"
curl "http://localhost:8000/async/items/1"

Update Item (using curl)

curl -X PUT "http://localhost:8000/sync/items/1" \
  -H "Content-Type: application/json" \
  -d '{"name": "Updated Laptop", "price": 1400}'

Delete Item (using curl)

curl -X DELETE "http://localhost:8000/sync/items/1"

🔍 Key Differences: Sync vs Async

Synchronous SQLAlchemy

# database_sync.py
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

# sync_router.py
def create_item_sync(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Characteristics:

• Blocking I/O operations
• Simpler to understand and debug
• Good for simple applications or CPU-bound tasks
• Uses psycopg2-binary driver

Asynchronous SQLAlchemy

# database_async.py
async_engine = create_async_engine(DATABASE_URL_ASYNC)
AsyncSessionLocal = async_sessionmaker(async_engine)

# async_router.py
async def create_item_async(item: ItemCreate, db: AsyncSession = Depends(get_async_db)):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

Characteristics:

• Non-blocking I/O operations
• Better performance under high concurrency
• More complex but more scalable
• Uses asyncpg driver
• Requires await for database operations

🛠 Tech Stack

• FastAPI - Modern, fast web framework
• SQLAlchemy 2.0 - SQL toolkit and ORM
• PostgreSQL - Relational database
• Pydantic - Data validation
• asyncpg - Async PostgreSQL driver
• psycopg2 - Sync PostgreSQL driver
• Docker - Containerization
• uv - Fast Python package installer

🎬 YouTube Video Script Ideas

1. Introduction (0:00-1:00)

   • Explain sync vs async concepts
   • Show project structure

2. Setup (1:00-3:00)

   • Docker Compose setup
   • Environment configuration
   • Installing dependencies with uv

3. Synchronous Implementation (3:00-7:00)

   • Database connection setup
   • Model definition
   • CRUD operations demo
   • Test with Swagger UI

4. Asynchronous Implementation (7:00-11:00)

   • Async engine setup
   • Key differences in code
   • CRUD operations demo
   • Performance comparison

5. Comparison (11:00-14:00)

   • Side-by-side code comparison
   • When to use sync vs async
   • Performance benchmarks

6. Conclusion (14:00-15:00)

   • Best practices
   • Recommendations
   • Resources

📊 Database Schema

Items Table:

Column	Type	Description
id	Integer	Primary key
name	String(255)	Item name
description	Text	Item description (optional)
price	Integer	Item price in cents
created_at	DateTime	Creation timestamp
updated_at	DateTime	Last update timestamp

🧪 Testing the Application

Using Swagger UI (Recommended)

1. Navigate to http://localhost:8000/docs
2. Try out the sync endpoints
3. Try out the async endpoints
4. Compare the responses

Using Python requests

import requests

# Create item
response = requests.post(
    "http://localhost:8000/sync/items",
    json={"name": "Mouse", "description": "Wireless mouse", "price": 50}
)
print(response.json())

# Get all items
response = requests.get("http://localhost:8000/sync/items")
print(response.json())

🐳 Docker Commands

# Start PostgreSQL
docker-compose up -d

# View logs
docker-compose logs -f

# Stop PostgreSQL
docker-compose down

# Stop and remove volumes
docker-compose down -v

🔧 Configuration

Edit .env file to customize:

POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres
POSTGRES_DB=testdb
POSTGRES_HOST=localhost
POSTGRES_PORT=5432

DATABASE_URL=postgresql://postgres:postgres@localhost:5432/testdb
DATABASE_URL_ASYNC=postgresql+asyncpg://postgres:postgres@localhost:5432/testdb

📈 Performance Tips

1. Use async for I/O-bound operations - Database queries, API calls
2. Use sync for CPU-bound operations - Heavy computations
3. Connection pooling - Both implementations use connection pools
4. Proper indexing - The name field is indexed for faster queries

🤝 Contributing

Feel free to fork, modify, and use this project for your learning and content creation!

📄 License

This project is open source and available for educational purposes.

🙏 Acknowledgments

• FastAPI documentation
• SQLAlchemy documentation
• Python async/await tutorials

---

Happy Coding! 🚀

If you found this helpful, please star ⭐ the repository!