What problem does it solve?
This skill addresses the challenge of establishing and maintaining reliable, high-performance asynchronous connections to Neon Serverless PostgreSQL from Python applications. It enables non-blocking data access and scalable ORM usage in serverless and microservice environments.
Core Features & Use Cases
- Async engine setup: Create an async engine with the proper driver and serverless-friendly pooling.
- Async session management: Provide a centralized
get_async_session pattern to ensure correct lifecycle management of sessions.
- Health checks & resilience: Implement lightweight health checks and connection health monitoring to detect stale or unhealthy connections.
- SQLModel + SQLAlchemy integration: Use SQLModel models with SQLAlchemy's Async API for consistent, scalable data access.
Quick Start
Install required packages: pip install sqlmodel sqlalchemy asyncpg
Set the environment variable DATABASE_URL to a Neon Serverless PostgreSQL URL using the postgresql+asyncpg:// scheme, for example:
postgresql+asyncpg://USER:PASSWORD@HOST/DB
Create an async engine with pool_pre_ping and an async session factory, then use async with blocks to manage sessions. Example setup steps:
- 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)
