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.