Setting Up PostgreSQL with FastAPI Asynchronously (Scalable Setup)

This guide walks you through setting up a professional and scalable integration between FastAPI and PostgreSQL using an async SQLAlchemy session manager. You'll learn how to configure the database, define models, and create API endpoints with proper session management.

---

Prerequisites

• Python 3.7 or higher installed.
• PostgreSQL installed locally or via Docker.
• Poetry installed:

  curl -sSL https://install.python-poetry.org | python3 -
  

---

Step 1: Set Up Your FastAPI Project

1.1 Create a New Project

poetry new fastapi-postgres
cd fastapi-postgres

1.2 Add Dependencies

Install FastAPI, Uvicorn, and async database tools:

poetry add fastapi uvicorn sqlalchemy databases asyncpg

---

Step 2: Configure PostgreSQL

2.1 Start a PostgreSQL Server

If using Docker:

docker run --name postgres -e POSTGRES_USER=fastapi -e POSTGRES_PASSWORD=password -p 5432:5432 -d postgres

2.2 Create a Database

Access the PostgreSQL shell:

docker exec -it postgres psql -U fastapi

Create a new database:

CREATE DATABASE fastapi_db;

---

Step 3: Configure the Database Connection

In config.py:

DATABASE_URL = "postgresql+asyncpg://fastapi:password@localhost:5432/fastapi_db"

---

Step 4: Set Up an Async SQLAlchemy Session Manager

In database.py:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from config import DATABASE_URL

# Create the database engine
engine = create_async_engine(DATABASE_URL, echo=True)

# Create an async session factory
async_sessionmaker = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# Dependency for getting a session
async def get_session() -> AsyncSession:
    async with async_sessionmaker() as session:
        yield session

---

Step 5: Define Models

In models.py:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)

---

Step 6: Create API Endpoints with Async Session Dependency

In main.py:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

from database import get_session, engine
from models import Base, Item

app = FastAPI()

@app.on_event("startup")
async def startup():
    # Create tables if they don’t exist
    async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

@app.post("/items/")
async def create_item(
    name: str,
    description: str,
    session: AsyncSession = Depends(get_session),
):
    new_item = Item(name=name, description=description)
    session.add(new_item)
    await session.commit()
    await session.refresh(new_item)
    return new_item

@app.get("/items/")
async def read_items(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Item))
    return result.scalars().all()

@app.get("/items/{item_id}")
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    if not item:
            raise HTTPException(status_code=404, detail="Item not found")
    return item

---

Step 7: Running Your Application

Run the application using Uvicorn:

poetry run uvicorn main:app --reload

Test the API:

1. Access http://127.0.0.1:8000/docs for interactive API documentation.
2. Test CRUD operations:
   • POST /items/: Create a new item.
   • GET /items/: List all items.
   • GET /items/{item_id}: Retrieve a specific item.

---

Wrapping Up

You’ve now set up a scalable and professional async session manager for your FastAPI project using SQLAlchemy. This approach ensures proper management of database connections in an asynchronous environment.

For more advanced features like migrations (using Alembic) or a production-ready boilerplate, check out Craft Your Startup. It’s the perfect way to kickstart your projects!