Neon DB Skill

Use this skill when working with Neon Serverless PostgreSQL databases.

Connection Pattern

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Neon connection string format
DATABASE_URL = "postgresql+asyncpg://user:password@ep-xyz.region.neon.tech/dbname?sslmode=require"

# Create async engine for Neon
engine = create_async_engine(
    DATABASE_URL,
    echo=True,  # Set to False in production
    pool_size=5,
    max_overflow=10,
)

# Session factory
async_session = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Environment Variables

# .env
DATABASE_URL="postgresql+asyncpg://user:pass@ep-xyz.region.neon.tech/db?sslmode=require"

Testing Connection

async def test_connection():
    """Test Neon database connection."""
    try:
        async with engine.begin() as conn:
            result = await conn.execute("SELECT 1")
            print("Connection successful!")
    except Exception as e:
        print(f"Connection failed: {e}")

Schema Operations

# Create table
async def create_tables():
    """Create all SQLModel tables."""
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# Drop table
async def drop_tables():
    """Drop all SQLModel tables."""
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)

Index Creation

async def create_indexes():
    """Create indexes for better query performance."""
    async with engine.begin() as conn:
        # Single column index
        await conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_tasks_user_id ON tasks(user_id)"
        )
        # Composite index
        await conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_user_completed ON tasks(user_id, completed)"
        )

Neon MCP Server Usage

Use @neon:run-sql for queries:

# Get tables
@neon:run_sql "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"

Use @neon:describe_table_schema for schema:

@neon:describe_table_schema tableName="tasks"

Best Practices

1. Always use sslmode=require for Neon
2. Use async engine with asyncpg driver
3. Set reasonable pool sizes (5-10)
4. Use expire_on_commit=False to avoid refresh issues
5. Always close sessions in finally block
6. Create indexes on foreign keys and frequently filtered columns
7. Use connection pooling for serverless functions