KAHIBARO
Discord Login Register ☰
☰

32.6. PostgreSQL Integration

Table of Contents

Overview

In the final project, you already know how to use PostgreSQL in general and how to use an ORM. This chapter focuses on how to integrate PostgreSQL into a real production‑style backend:

We will keep examples Python oriented (FastAPI + SQLAlchemy), but the ideas apply to any backend stack.

Goal of this chapter:
Connect your application to PostgreSQL in a way that is secure, configurable, and production ready, not just "it works on my machine."

Defining PostgreSQL configuration

In a production backend, you never hardcode database credentials in source code. You keep them in environment variables or a secrets store, and your application builds a connection URL from them.

Environment variables for PostgreSQL

A typical set of environment variables for PostgreSQL:

VariableExample valueDescription
DB_HOSTdb or 127.0.0.1Database hostname
DB_PORT5432PostgreSQL port
DB_NAMEmyappDatabase name
DB_USERmyapp_userUsername
DB_PASSWORDsuper-secretPassword
DB_POOL_SIZE10Max connections per worker
DB_POOL_MAX_OVERFLOW20Extra temporary connections
DB_POOL_TIMEOUT30Seconds to wait for a connection

In development you might have a .env file:

env 
DB_HOST=localhost
DB_PORT=5432
DB_NAME=myapp_dev
DB_USER=myapp_dev
DB_PASSWORD=devpassword
DB_POOL_SIZE=5
DB_POOL_MAX_OVERFLOW=10
DB_POOL_TIMEOUT=30

In production you set these through your orchestrator or server (Docker Compose, Kubernetes, systemd unit, etc.).

Rule: Database credentials must not live in source control.
Use environment variables, secret managers, or configuration management tools.

Building the database URL

Most libraries use a single URL string, for example:

text 
postgresql+psycopg2://user:password@host:5432/dbname

In Python:

python 
import os
from urllib.parse import quote_plus
def build_db_url() -> str:
    user = os.getenv("DB_USER", "postgres")
    password = os.getenv("DB_PASSWORD", "")
    host = os.getenv("DB_HOST", "localhost")
    port = os.getenv("DB_PORT", "5432")
    name = os.getenv("DB_NAME", "postgres")
    # URL-encode password in case it has special characters
    password_escaped = quote_plus(password)
    return f"postgresql+psycopg2://{user}:{password_escaped}@{host}:{port}/{name}"
DATABASE_URL = build_db_url()

You can also expose DATABASE_URL directly as an environment variable and skip building it yourself, but building it from smaller variables often makes it easier to change parts per environment.

Creating the database engine and session

Your ORM or driver needs a central place where you define:

SQLAlchemy engine and session factory

A common pattern in a FastAPI project:

python 
# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
import os
class Base(DeclarativeBase):
    pass
DATABASE_URL = os.getenv("DATABASE_URL")
engine = create_engine(
    DATABASE_URL,
    pool_size=int(os.getenv("DB_POOL_SIZE", "10")),
    max_overflow=int(os.getenv("DB_POOL_MAX_OVERFLOW", "20")),
    pool_timeout=int(os.getenv("DB_POOL_TIMEOUT", "30")),
    pool_pre_ping=True,  # Check connections before using them
)
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)

Key parameters:

ParameterMeaning
pool_sizeNumber of persistent connections kept in the pool
max_overflowExtra connections that can be created when pool is full
pool_timeoutHow long to wait for a connection from the pool before raising an error
pool_pre_pingChecks that a connection is still alive, prevents "stale connection" errors

For most APIs, a small pool per worker like pool_size=5 is enough.

Rule: Always use connection pooling in production.
Opening and closing a PostgreSQL connection on every request is too slow and wastes resources.

FastAPI dependency for database sessions

You want each HTTP request to use its own database session, and you want to make sure the session is closed, even when an error happens.

python 
# app/dependencies/database.py
from typing import Generator
from app.db import SessionLocal
def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

Using it in a route:

python 
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.dependencies.database import get_db
from app import models, schemas
router = APIRouter()
@router.get("/users/{user_id}", response_model=schemas.UserRead)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        # Raise your custom HTTP 404 here
        ...
    return user

Every request:

  1. Opens a session.
  2. Runs the route code.
  3. Commits on success, rolls back on error.
  4. Closes the session.

Managing database migrations

In a real project, you never manually change schema on production. You use migrations so schema changes are:

This project uses Alembic (mentioned in the ORM chapter). Here we only show how it fits into the final project.

Basic migration workflow in the project

Typical layout:

text 
app/
  db.py
  models/
    __init__.py
    user.py
    product.py
  ...
alembic/
  versions/
  env.py
alembic.ini

Once you have models defined and the Alembic setup created:

  1. Generate a new migration:
bash 
   alembic revision --autogenerate -m "create users and products tables"
  1. Inspect the generated file and confirm it looks correct.
  2. Apply migrations to your local database:
bash 
   alembic upgrade head
  1. In CI/CD or deployment, run:
bash 
   alembic upgrade head

right after deploying the new version of the application.

Rule: Never edit production schemas by hand.
All schema changes must go through migrations and version control.

Environment specific migration configuration

You want migrations to run against the right database:

In alembic/env.py you usually read the same DATABASE_URL as the app, or a special DB_URL_MIGRATIONS variable.

Example snippet:

python 
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
import os
from app.db import Base  # to include models' metadata
config = context.config
# Override from env var if set
database_url = os.getenv("DATABASE_URL")
if database_url:
    config.set_main_option("sqlalchemy.url", database_url)
if config.config_file_name is not None:
    fileConfig(config.config_file_name)
target_metadata = Base.metadata

This way you can run:

bash 
DATABASE_URL=postgresql+psycopg2://user:pass@host/prod_db alembic upgrade head

and Alembic will migrate the production database.

Handling multiple environments

The same application code often runs with 3 configurations:

They all use PostgreSQL, but with different DBs and stricter or looser settings.

Separate databases per environment

A simple naming convention:

EnvironmentExample DB name
Devmyapp_dev
Testmyapp_test
Prodmyapp

You can control this with an APP_ENV variable:

env 
APP_ENV=development
DB_NAME=myapp_dev

In production:

env 
APP_ENV=production
DB_NAME=myapp

Or you expose different DATABASE_URL values entirely.

Read-only vs read-write access

Sometimes you will want:

In a full-scale system you might configure two connection URLs:

python 
DATABASE_URL_RW = os.getenv("DATABASE_URL_RW")
DATABASE_URL_RO = os.getenv("DATABASE_URL_RO")  # user with SELECT only

In this final project you can keep one main URL, but it is good to know how to extend the design later.

Health checks and connectivity

In production, your app must know whether it can talk to PostgreSQL. This is useful for:

Simple database health check endpoint

You can create a minimal endpoint that asks PostgreSQL for a trivial query, for example SELECT 1.

python 
# app/api/health.py
from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session
from app.dependencies.database import get_db
router = APIRouter()
@router.get("/health/db", status_code=status.HTTP_200_OK)
def db_health(db: Session = Depends(get_db)):
    # Simple query to check connection
    db.execute("SELECT 1")
    return {"status": "ok"}

You can then configure your load balancer or Kubernetes to hit /health/db.

Rule: A health check should test real dependencies.
For database health, actually run a small SQL query.

Connection pooling and concurrency

In a production backend, you usually have:

Matching pool size to PostgreSQL limits

PostgreSQL has a max_connections setting. Each worker in your app creates its own pool. You must ensure the total pooled connections do not exceed what PostgreSQL can handle.

If you run:

Then you can have up to $4 \times 5 = 20$ open connections, plus overflows.

You should choose pool_size such that:

$$
\text{Number of workers} \times \text{pool\_size} \ll \text{max\_connections}
$$

If PostgreSQL has max_connections=100 and you reserve some for admin, a safe configuration might be:

So at worst: $4 \times (10 + 10) = 80$ connections.

Pre-ping and stale connections

Long-running APIs sometimes lose connections if:

pool_pre_ping=True solves many of these problems:

python 
engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    # other params...
)

It runs a simple "ping" on a connection before it is reused. If it is broken, SQLAlchemy discards it and opens a new one.

Structuring code for database access

You already met the Repository Pattern in a previous chapter. Here is how it might look inside the final project, with PostgreSQL behind the scenes.

Example project layout with repositories

text 
app/
  db.py
  models/
    __init__.py
    user.py
  repositories/
    __init__.py
    user_repository.py
  schemas/
    __init__.py
    user.py
  api/
    __init__.py
    users.py

User model example

python 
# app/models/user.py
from sqlalchemy import Column, Integer, String
from app.db import Base
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)

User repository example

python 
# app/repositories/user_repository.py
from typing import Optional, List
from sqlalchemy.orm import Session
from app.models.user import User
class UserRepository:
    def __init__(self, db: Session):
        self.db = db
    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.db.query(User).filter(User.id == user_id).first()
    def get_by_email(self, email: str) -> Optional[User]:
        return self.db.query(User).filter(User.email == email).first()
    def list(self, limit: int = 100, offset: int = 0) -> List[User]:
        return (
            self.db.query(User)
            .order_by(User.id)
            .offset(offset)
            .limit(limit)
            .all()
        )
    def create(self, email: str, hashed_password: str) -> User:
        user = User(email=email, hashed_password=hashed_password)
        self.db.add(user)
        # commit is handled in get_db dependency
        self.db.flush()  # To get the ID without commit
        self.db.refresh(user)
        return user

Using repository in a route

python 
# app/api/users.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.dependencies.database import get_db
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate, UserRead
router = APIRouter(prefix="/users", tags=["users"])
@router.post("", response_model=UserRead)
def create_user(
    payload: UserCreate,
    db: Session = Depends(get_db),
):
    repo = UserRepository(db)
    existing = repo.get_by_email(payload.email)
    if existing:
        # raise HTTP 400: email already exists
        ...
    user = repo.create(
        email=payload.email,
        hashed_password=payload.hashed_password,  # hashed in service layer
    )
    return user

Here PostgreSQL is hidden behind both SQLAlchemy and a repository interface. This makes it easy to swap PostgreSQL for another database in the future or to mock the repository in tests.

Using PostgreSQL in tests

For your final project, you should have automated tests that use a separate PostgreSQL database or an in-memory / ephemeral DB.

Strategy 1: Real PostgreSQL test database

Steps:

  1. Create myapp_test database.
  2. Configure DATABASE_URL for tests:
env 
   TEST_DATABASE_URL=postgresql+psycopg2://myapp_test:password@localhost/myapp_test
  1. In your test configuration, build an engine against TEST_DATABASE_URL.
  2. Create all tables at test startup.
  3. Use transactions or truncation between tests.

Example pytest fixture:

python 
# tests/conftest.py
import os
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.db import Base
from app.main import app
from fastapi.testclient import TestClient
TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL")
engine = create_engine(TEST_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="session", autouse=True)
def setup_database():
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)
@pytest.fixture
def db_session():
    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.close()
@pytest.fixture
def client(db_session, monkeypatch):
    from app.dependencies import database as db_dep
    def override_get_db():
        try:
            yield db_session
            db_session.commit()
        except Exception:
            db_session.rollback()
            raise
    app.dependency_overrides[db_dep.get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

Now your tests talk to a real PostgreSQL instance and you can test your integration properly.

Docker and PostgreSQL for the final project

Many production deployments and local setups will use Docker for both the app and the database.

Example docker-compose.yml snippet

yaml 
version: "3.9"
services:
  db:
    image: postgres:16
    restart: unless-stopped
    environment:
      POSTGRES_DB: myapp
      POSTGRES_USER: myapp_user
      POSTGRES_PASSWORD: super-secret
    ports:
      - "5432:5432"
    volumes:
      - myapp_pgdata:/var/lib/postgresql/data
  api:
    build: ./app
    depends_on:
      - db
    environment:
      DB_HOST: db
      DB_PORT: 5432
      DB_NAME: myapp
      DB_USER: myapp_user
      DB_PASSWORD: super-secret
    ports:
      - "8000:8000"
volumes:
  myapp_pgdata:

Notice:

You can run migrations inside the api container:

bash 
docker compose run --rm api alembic upgrade head

or use a dedicated migrations service.

Basic production hardening

Before you ship the final project, check some PostgreSQL related security and reliability practices.

Database user and permissions

Create a dedicated application user with only the privileges it needs:

  1. Log in as a superuser:
sql 
   CREATE USER myapp_user WITH PASSWORD 'super-secret';
   CREATE DATABASE myapp OWNER myapp_user;
  1. Restrict direct superuser access in production as far as possible.
  2. Avoid connecting as postgres from the app.

SSL / TLS to PostgreSQL

In many hosting providers, connections to the database are encrypted. You may need to add parameters like:

text 
postgresql+psycopg2://user:pass@host/dbname?sslmode=require

Check your provider's documentation for the recommended sslmode and certificates.

Regular backups

The "Database Backups" chapter covers this in detail. For the final project:

Checklist for PostgreSQL integration in the final project

Use this as a final self review before you move on:

If you can check all of these items, your PostgreSQL integration is in good shape for a real production backend.

Views: 7

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3096 / 24h
Impressum / Imprint
Datenschutz / Data Policy