Neon PostgreSQL Async Connection

This skill providing guidance on setting up and managing reliable asynchronous database connections for Neon Serverless PostgreSQL.

Purpose

Setting up and managing reliable, high-performance asynchronous database connections for Neon Serverless PostgreSQL using SQLModel and SQLAlchemy's async capabilities.

Capabilities

• Environment-aware configuration using DATABASE_URL.
• Creating an asynchronous SQLAlchemy engine using the asyncpg driver.
• Implementing a get_async_session dependency utilizing proper async context management.
• Centralized Base metadata management for SQLModel tables.
• Connection health check utilities for operational monitoring.
• Serverless-optimized connection pooling configuration.

Implementation Details

Driver & Engine

Ensuring the DATABASE_URL uses the postgresql+asyncpg:// scheme.

from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.orm import sessionmaker

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

async def get_async_session():
    async with async_session() as session:
        yield session

Health Check

Implementing a lightweight query to verify connectivity.

async def check_db_health():
    async with engine.begin() as conn:
        await conn.execute(text("SELECT 1"))

Best Practices

• Using pool_pre_ping=True to handle stale connections commonly found in serverless environments.
• Setting expire_on_commit=False in async sessions to allow access to object attributes after a commit.
• Always using async context managers (async with) for sessions to ensure they are properly returned to the pool.