How to Connect to PostgreSQL with FastAPI: Simple Guide

To connect FastAPI to PostgreSQL, use an async database library like SQLAlchemy with asyncpg driver or the databases package. Define your database URL, create an async engine, and use dependency injection to access the database in your routes.

Syntax

This is the basic pattern to connect FastAPI to PostgreSQL using async SQLAlchemy and the asyncpg driver:

• DATABASE_URL: Your PostgreSQL connection string.
• create_async_engine: Creates an async engine for database communication.
• AsyncSession: Manages async database sessions.
• Depends: Injects the session into FastAPI routes.

python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

app = FastAPI()

Example

This example shows a complete FastAPI app connecting to PostgreSQL, creating a simple table, and querying data asynchronously.

python

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

app = FastAPI()

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/users/")
async def create_user(name: str, session: AsyncSession = Depends(get_session)):
    user = User(name=name)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return {"id": user.id, "name": user.name}

@app.get("/users/")
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

Output

When running, the app connects to PostgreSQL, creates the users table, and supports POST /users/ to add users and GET /users/ to list them.

Common Pitfalls

Common mistakes when connecting FastAPI to PostgreSQL include:

• Using synchronous database drivers instead of async ones, causing blocking.
• Not using async_session and AsyncSession for async operations.
• Forgetting to create tables before querying, leading to errors.
• Hardcoding credentials without environment variables, risking security.

python

## Wrong: Using synchronous engine in async FastAPI
from sqlalchemy import create_engine
engine = create_engine("postgresql://user:password@localhost/dbname")  # sync engine

## Right: Use async engine with asyncpg
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")

Quick Reference

Tips for connecting FastAPI to PostgreSQL:

• Use postgresql+asyncpg in your database URL for async support.
• Use AsyncSession and async_session for database sessions.
• Run Base.metadata.create_all on startup to create tables.
• Use dependency injection with Depends to get sessions in routes.
• Keep credentials secure using environment variables or secrets management.

Key Takeaways

Use async SQLAlchemy with the asyncpg driver for non-blocking PostgreSQL access in FastAPI.

Create an async engine and sessionmaker, then inject sessions into routes with Depends.

Always create your database tables on app startup to avoid runtime errors.

Avoid synchronous database drivers to prevent blocking FastAPI's async event loop.

Secure your database credentials and avoid hardcoding them in your code.