db-connection

Database Connection Skill

Expert database connection management for Python/FastAPI with Neon PostgreSQL, connection pooling, and SSL configuration.

Quick Reference

| Task | File/Method | |------|-------------| | Get engine | get_engine() | | Get session | get_session() | | Connection string | DB_URL from settings | | Health check | check_connection() |

Project Structure

backend/
├── app/
│   ├── db/
│   │   ├── __init__.py
│   │   ├── connection.py    # Engine and session setup
│   │   └── session.py       # Dependency injection
│   └── config/
│       └── settings.py      # Environment config
├── alembic/
│   └── env.py               # Uses connection from here
└── .env.example

Connection Configuration

Settings with DB URL

# backend/app/config/settings.py
from functools import lru_cache
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # Database Configuration
    DB_URL: SecretStr = Field(
        ...,
        description="PostgreSQL connection URL",
        examples=["postgresql://user:pass@ep-xxx.us-east-1.aws.neon.tech/dbname?sslmode=require"],
    )
    DB_POOL_SIZE: int = Field(default=5, ge=1, le=100)
    DB_MAX_OVERFLOW: int = Field(default=10, ge=0, le=100)
    DB_POOL_TIMEOUT: int = Field(default=30, ge=1, le=300)
    DB_POOL_RECYCLE: int = Field(default=1800, ge=300)
    DB_ECHO: bool = False


@lru_cache
def get_settings() -> Settings:
    return Settings()

Environment Variables

# .env.example

# Database (Neon PostgreSQL)
# Get this from Neon Dashboard > Connection Details
# Format: postgresql://user:pass@host/dbname?sslmode=require
DB_URL="postgresql://username:password@ep-xxx.region.neon.tech/dbname?sslmode=require"

# Connection Pool Settings
DB_POOL_SIZE=5
DB_MAX_OVERFLOW=10
DB_POOL_TIMEOUT=30
DB_POOL_RECYCLE=1800

# Debug (set to true for development)
DB_ECHO=false

SQLAlchemy Engine Setup

Connection Module

# backend/app/db/connection.py
from sqlalchemy import create_engine, event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator
import logging

from app.config.settings import get_settings

logger = logging.getLogger(__name__)


def get_db_url() -> str:
    """Get database URL from settings."""
    settings = get_settings()
    db_url = settings.DB_URL
    # SecretStr has get_secret_value() method
    if hasattr(db_url, 'get_secret_value'):
        return db_url.get_secret_value()
    return str(db_url)


def create_sqlalchemy_engine() -> Engine:
    """Create SQLAlchemy engine with optimal settings for Neon/PostgreSQL."""
    settings = get_settings()
    db_url = get_db_url()

    engine = create_engine(
        db_url,
        pool_size=settings.DB_POOL_SIZE,
        max_overflow=settings.DB_MAX_OVERFLOW,
        pool_timeout=settings.DB_POOL_TIMEOUT,
        pool_recycle=settings.DB_POOL_RECYCLE,
        echo=settings.DB_ECHO,
        # PostgreSQL-specific settings
        pool_pre_ping=True,  # Verify connections before use
        isolation_level="AUTOCOMMIT",
    )

    # Enable connection health checks
    @event.listens_for(engine, "connect")
    def set_session_vars(dbapi_connection, connection_record):
        cursor = dbapi_connection.cursor()
        # Set session characteristics
        cursor.execute("SET statement_timeout = '30s'")
        cursor.execute("SET idle_in_transaction_session_timeout = '60000'")
        cursor.close()

    logger.info(f"Database engine created with pool_size={settings.DB_POOL_SIZE}")
    return engine


def get_engine() -> Engine:
    """Get or create database engine (singleton pattern)."""
    return create_sqlalchemy_engine()

Session Management

# backend/app/db/session.py
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator
from app.db.connection import get_engine


# Create session factory
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=get_engine(),
)


def get_db() -> Generator[Session, None, None]:
    """
    Database session dependency for FastAPI.

    Usage:
        @router.get("/users/")
        def get_users(db: Session = Depends(get_db)):
            ...
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


async def get_async_db() -> Generator[Session, None, None]:
    """
    Async database session dependency (for async routes).
    Note: Use with SQLModel async sessions or asyncpg.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Neon PostgreSQL Setup

Neon Connection String Format

postgresql://[user]:[password]@[host]/[dbname]?sslmode=require

Components:

• user: Database username (from Neon)
• password: Database password (from Neon)
• host: Endpoint ID + region, e.g., ep-xxx-12345.us-east-1.aws.neon.tech
• dbname: Your database name
• sslmode: Must be require for Neon

Getting Connection Details from Neon

1. Go to Neon Dashboard
2. Select your project
3. Go to Connection Details
4. Copy the connection string
5. Add to Vercel/Dashboard environment variables

Connection Pooling for Neon

# For serverless/edge functions, use lower pool sizes
# backend/app/db/connection.py

def create_serverless_engine() -> Engine:
    """Create engine optimized for serverless/Vercel functions."""
    settings = get_settings()

    # Smaller pool for serverless to avoid connection limits
    return create_engine(
        get_db_url(),
        pool_size=2,  # Keep small for serverless
        max_overflow=0,  # No overflow in serverless
        pool_timeout=10,  # Faster timeout
        pool_recycle=300,  # Recycle more frequently
        pool_pre_ping=True,
        echo=settings.DB_ECHO,
    )

FastAPI Integration

Application Setup

# backend/app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db.connection import get_engine
from app.db.session import get_db
from app.config.settings import get_settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Verify database connection
    settings = get_settings()
    engine = get_engine()

    try:
        with engine.connect() as conn:
            conn.execute("SELECT 1")
        logger.info("Database connection verified successfully")
    except Exception as e:
        logger.error(f"Database connection failed: {e}")
        raise

    yield

    # Shutdown: Close all connections
    engine.dispose()
    logger.info("Database connections closed")


app = FastAPI(lifespan=lifespan)


# Dependency injection works with any route
@app.get("/users/")
def get_users(db=Depends(get_db)):
    return db.query(User).all()

Database Health Check

# backend/app/api/health.py
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.db.session import get_db

router = APIRouter()


@router.get("/health/db")
def database_health(db: Session = Depends(get_db)) -> dict:
    """
    Check database connectivity.

    Returns:
        {
            "status": "healthy",
            "latency_ms": <response_time>,
            "database": <db_name>
        }
    """
    import time

    start = time.time()
    result = db.execute(text("SELECT 1"))
    latency_ms = (time.time() - start) * 1000

    return {
        "status": "healthy",
        "latency_ms": round(latency_ms, 2),
        "database": "postgresql",
    }

SSL Configuration

Required SSL Settings

# Neon requires SSL - this is the default behavior
# No additional configuration needed when using ?sslmode=require

# Verify