Using Relational Databases

Relational Databases

Purpose

This skill guides relational database selection and implementation across multiple languages. Choose the optimal database engine, ORM/query builder, and deployment strategy for transactional systems, CRUD applications, and structured data storage.

When to Use This Skill

Trigger this skill when:

• Building user authentication, content management, e-commerce applications
• Implementing CRUD operations (Create, Read, Update, Delete)
• Designing data models with relationships (users → posts, orders → items)
• Migrating schemas safely in production
• Setting up connection pooling for performance
• Evaluating serverless database options (Neon, PlanetScale, Turso)
• Integrating with frontend skills (forms, tables, dashboards, search-filter)

Skip this skill for:

• Time-series data at scale (use time-series databases)
• Real-time analytics (use columnar databases)
• Document-heavy workloads (use document databases)
• Key-value caching (use Redis, Memcached)

Quick Reference: Database Selection

Database Selection Decision Tree
═══════════════════════════════════════════════════════════

PRIMARY CONCERN?
├─ MAXIMUM FLEXIBILITY & EXTENSIONS (JSON, arrays, vector search)
│  └─ PostgreSQL
│     ├─ Serverless → Neon (scale-to-zero, database branching)
│     └─ Traditional → Self-hosted, AWS RDS, Google Cloud SQL
│
├─ EMBEDDED / EDGE DEPLOYMENT (local-first, global latency)
│  └─ SQLite or Turso
│     ├─ Global distribution → Turso (libSQL, edge replicas)
│     └─ Local-only → SQLite (embedded, zero-config)
│
├─ LEGACY SYSTEM / MYSQL REQUIRED
│  └─ MySQL
│     ├─ Serverless → PlanetScale (non-blocking migrations)
│     └─ Traditional → Self-hosted, AWS RDS, Google Cloud SQL
│
└─ RAPID PROTOTYPING
   ├─ Python → SQLModel (FastAPI) or SQLAlchemy 2.0
   ├─ TypeScript → Prisma (best DX) or Drizzle (performance)
   ├─ Rust → SQLx (compile-time checks)
   └─ Go → sqlc (type-safe code generation)

Quick Reference: ORM vs Query Builder

ORM vs Query Builder Selection
═══════════════════════════════════════════════════════════

TEAM PRIORITIES?
├─ DEVELOPMENT SPEED / DEVELOPER EXPERIENCE
│  └─ ORM (abstracts SQL, handles relations automatically)
│     ├─ Python → SQLAlchemy 2.0, SQLModel
│     ├─ TypeScript → Prisma (migrations, type generation)
│     ├─ Rust → SeaORM (Active Record + Data Mapper)
│     └─ Go → GORM, Ent
│
├─ PERFORMANCE / QUERY CONTROL
│  └─ Query Builder (SQL-like, zero abstraction overhead)
│     ├─ Python → SQLAlchemy Core, asyncpg
│     ├─ TypeScript → Drizzle, Kysely
│     ├─ Rust → SQLx (compile-time query validation!)
│     └─ Go → sqlc (generates types from SQL)
│
├─ TYPE SAFETY / COMPILE-TIME GUARANTEES
│  ├─ Rust → SQLx (queries checked at build time)
│  ├─ Go → sqlc (generates types from SQL)
│  ├─ TypeScript → Prisma or Drizzle
│  └─ Python → SQLModel (Pydantic integration)
│
└─ COMPLEX QUERIES / JOINS
   ├─ SQL-first → Query builders or raw SQL
   └─ ORM-friendly → SeaORM, SQLAlchemy ORM

Multi-Language Implementation

Python: SQLAlchemy 2.0 + SQLModel

Recommended Libraries:

• SQLAlchemy 2.0 (/websites/sqlalchemy_en_21) - ORM + Core, 7,090 snippets
• SQLModel - FastAPI integration, Pydantic validation
• asyncpg - High-performance async PostgreSQL driver

When to Use:

• Production applications requiring flexibility
• FastAPI/Starlette backends
• Async/await workflows

Quick Pattern:

from sqlmodel import SQLModel, Field, Session
class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)

See: references/orms-python.md for complete SQLAlchemy/SQLModel patterns, async workflows, and connection pooling.

TypeScript: Prisma vs Drizzle

Recommended Libraries:

• Prisma 6.x (/prisma/prisma, score: 96.4, 4,281 doc snippets) - Best DX, migrations
• Drizzle ORM (/drizzle-team/drizzle-orm-docs, score: 95.4, 4,037 snippets) - Performance, SQL-like

Quick Comparison:

• Prisma: Best DX, auto-generated types, migrations included
• Drizzle: Best performance, SQL-like syntax, zero overhead

See: references/orms-typescript.md for Prisma vs Drizzle detailed comparison, Kysely, TypeORM patterns.

Rust: SQLx (Compile-Time Checked)

Recommended Libraries:

• SQLx 0.8 - Compile-time query validation, async
• SeaORM 1.x - Full ORM with Active Record pattern
• Diesel 2.3 - Mature, stable (sync/async)

Quick Pattern:

use sqlx::FromRow;
#[derive(FromRow)]
struct User { id: i32, email: String, name: String }
// Compile-time checked queries (verified at build time!)
let user = sqlx::query_as::<_, User>("SELECT * FROM users WHERE email = $1")
    .bind("test@example.com").fetch_one(&pool).await?;

See: references/orms-rust.md for SQLx macros, SeaORM, Diesel patterns, and compile-time guarantees.

Go: sqlc (Type-Safe Code Generation)

Recommended Libraries:

• sqlc - Generates Go code from SQL queries
• GORM v2 - Full ORM with associations, hooks
• Ent - Graph-based ORM, schema as code
• pgx - High-performance PostgreSQL driver

Quick Pattern:

-- queries.sql: SQL annotations generate type-safe Go code
-- name: CreateUser :one
INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *;

user, err := queries.CreateUser(ctx, db.CreateUserParams{Email: "test@example.com"})

See: references/orms-go.md for sqlc setup, GORM, Ent, and pgx patterns.

Connection Pooling

Recommended Pool Sizes:

• Web API (single instance): 10-20 connections
• Serverless (per function): 1-2 connections + pgBouncer
• Background workers: 5-10 connections

See: references/connection-pooling.md for configuration examples, sizing formulas, and monitoring strategies.

Migrations

Critical Principles: 1. Use multi-phase deployment for column drops (never drop directly in production) 2. Use CREATE INDEX CONCURRENTLY (PostgreSQL) to avoid blocking writes 3. Test migrations in staging with production-like data volume

Tools: Alembic (Python), Prisma Migrate (TypeScript), SQLx migrations (Rust), golang-migrate (Go)

See: references/migrations-guide.md for safe migration patterns, multi-phase deployments, and rollback strategies.

Serverless Databases

See: references/serverless-databases.md for setup examples, branching workflows, and cost comparisons.

Frontend Integration

Common Integration Patterns:

• Forms skill: Form submission → API validation → Database CRUD (INSERT/UPDATE)
• Tables skill: Paginated queries → API → Table display with sorting/filtering
• Dashboards skill: Aggregation queries (COUNT, SUM) → API → KPI cards
• Search-filter skill: Full-text search (PostgreSQL tsvector) → Ranked results

See working examples in: examples/python-sqlalchemy/, examples/typescript-drizzle/, examples/rust-sqlx/

Bundled Resources

Reference Documentation

• references/postgresql-guide.md - PostgreSQL features (pgvector, PostGIS, TimescaleDB)
• references/mysql-guide.md - MySQL-specific patterns, PlanetScale integration
• references/sqlite-guide.md - SQLite patterns, Turso edge deployment
• references/orms-python.md - SQLAlchemy 2.0, SQLModel, asyncpg
• references/orms-typescript.md - Prisma, Drizzle, Kysely comparisons
• references/orms-rust.md - SQLx, SeaORM, Diesel
• references/orms-go.md - GORM, sqlc, Ent, pgx
• references/migrations-guide.md - Safe schema evolution patterns
• references/connection-pooling.md - Pool sizing and monitoring
• references/serverless-databases.md - Neon, PlanetScale, Turso deployment

Working Examples

• examples/python-sqlalchemy/ - SQLAlchemy 2.0 + FastAPI with pooling, migrations
• examples/typescript-prisma/ - Prisma + Next.js with schema, migrations
• examples/typescript-drizzle/ - Drizzle + Hono with type-safe queries
• examples/rust-sqlx/ - SQLx + Axum with compile-time checks
• examples/go-sqlc/ - sqlc + Gin with generated type-safe code

Utility Scripts

• scripts/validate_schema.py - Validate database schema structure, constraints
• scripts/generate_migration.py - Generate migration templates for common operations

Best Practices

Security:

• Always use parameterized queries (prevents SQL injection)
• Hash passwords with Argon2/bcrypt
• Use environment variables for connection strings
• Enable SSL/TLS in production

Performance:

• Use connection pooling (10-20 for web APIs)
• Create indexes on filtered/sorted columns
• Implement pagination for large result sets
• Use EXPLAIN ANALYZE for slow queries

Reliability:

• Test migrations in staging first
• Use transactions for multi-statement operations
• Monitor connection pool exhaustion
• Set up and test database backups

Development:

• Version control schema and migrations
• Use database branching (Neon) for features
• Write integration tests against real databases

Python SQLAlchemy Example

Production-ready FastAPI + SQLAlchemy 2.0 + PostgreSQL example with connection pooling and migrations.

Features

• SQLAlchemy 2.0 with async support
• FastAPI REST API
• Connection pooling configuration
• Alembic migrations
• User CRUD operations
• Password hashing with Argon2

Setup

Prerequisites

• Python 3.11+
• PostgreSQL 14+

Installation

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

Database Setup

# Set database URL
export DATABASE_URL="postgresql://user:password@localhost:5432/mydb"

# Run migrations
alembic upgrade head

Running

# Development server with auto-reload
uvicorn main:app --reload

# Production server
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

API Endpoints

Create User

curl -X POST http://localhost:8000/users \
  -H "Content-Type: application/json" \
  -d '{"email": "test@example.com", "name": "Test User", "password": "securepassword"}'

Get User

curl http://localhost:8000/users/1

List Users

curl "http://localhost:8000/users?skip=0&limit=10"

Update User

curl -X PUT http://localhost:8000/users/1 \
  -H "Content-Type: application/json" \
  -d '{"name": "Updated Name"}'

Delete User

curl -X DELETE http://localhost:8000/users/1

Connection Pooling

Configured in main.py:

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,           # 20 normal connections
    max_overflow=10,        # 10 extra under load (total max = 30)
    pool_pre_ping=True,     # Verify connections before use
    pool_recycle=3600,      # Recycle after 1 hour
)

Tuning:

• Web API: pool_size=20, max_overflow=10
• Background worker: pool_size=5, max_overflow=5
• Serverless: pool_size=1, max_overflow=0 (use pgBouncer)

Migrations

# Create new migration
alembic revision --autogenerate -m "Add users table"

# Apply migrations
alembic upgrade head

# Rollback one migration
alembic downgrade -1

# View migration history
alembic history

Project Structure

python-sqlalchemy/
├── alembic/
│   └── versions/
│       └── 001_initial.py
├── models.py          # Database models
├── schemas.py         # Pydantic schemas
├── main.py            # FastAPI application
├── requirements.txt   # Python dependencies
└── README.md

Production Checklist

• [ ] Use environment variables for DATABASE_URL
• [ ] Enable SSL for database connections
• [ ] Set up database backups
• [ ] Configure logging
• [ ] Add rate limiting
• [ ] Implement authentication/authorization
• [ ] Set up monitoring (prometheus, datadog)
• [ ] Use pgBouncer for connection pooling

Connection Pooling Guide

Table of Contents

• Why Connection Pooling Matters
• Recommended Pool Sizes
• Configuration by Language
• Python (SQLAlchemy)
• TypeScript (Prisma)
• TypeScript (Drizzle + pg)
• Rust (SQLx)
• Go (database/sql)
• Go (pgx)
• Serverless Considerations
• Problem
• Solution: pgBouncer
• Serverless Database Services
• Monitoring Connection Pools
• PostgreSQL
• Application Metrics
• Best Practices
• 1. Right-Size Your Pool
• 2. Use Connection Lifecycle Management
• 3. Handle Connection Errors
• 4. Close Connections Properly
• 5. Serverless: Use Connection Poolers
• Troubleshooting
• "Too many connections" Error
• High Connection Wait Times
• Stale Connections
• Resources

Why Connection Pooling Matters

Creating database connections is expensive:

• TCP handshake: 50-100ms
• Authentication: 50-200ms
• Total: 100-500ms per connection

Connection pooling reuses connections, reducing latency from 100ms to <1ms.

Recommended Pool Sizes

Formula: pool_size = (num_cores * 2) + num_disks

• Example: 4 cores, 1 disk = 10 connections
• More connections != better (causes contention)

Configuration by Language

Python (SQLAlchemy)

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,           # Normal connections
    max_overflow=10,        # Extra connections under load (total max = 30)
    pool_timeout=30,        # Wait 30s for connection before error
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True      # Verify connection before use (prevents stale connections)
)

TypeScript (Prisma)

// Configure in DATABASE_URL
// postgresql://user:pass@host/db?connection_limit=20&pool_timeout=30

// Or programmatically
const prisma = new PrismaClient({
  datasources: {
    db: {
      url: process.env.DATABASE_URL,
    },
  },
})

TypeScript (Drizzle + pg)

import { Pool } from 'pg'

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 20,                    // Max connections
  min: 5,                     // Min idle connections
  idleTimeoutMillis: 30000,   // Close idle after 30s
  connectionTimeoutMillis: 2000,  // Wait 2s for connection
})

Rust (SQLx)

use sqlx::postgres::PgPoolOptions;
use std::time::Duration;

let pool = PgPoolOptions::new()
    .max_connections(20)
    .min_connections(5)
    .acquire_timeout(Duration::from_secs(30))
    .idle_timeout(Some(Duration::from_secs(600)))  // 10 min
    .max_lifetime(Some(Duration::from_secs(3600))) // 1 hour
    .connect("postgresql://...")
    .await?;

Go (database/sql)

import "database/sql"

db, _ := sql.Open("postgres", "postgresql://...")
db.SetMaxOpenConns(20)                    // Max connections
db.SetMaxIdleConns(10)                    // Max idle connections
db.SetConnMaxLifetime(time.Hour)          // Max lifetime
db.SetConnMaxIdleTime(time.Minute * 10)   // Max idle time

Go (pgx)

import "github.com/jackc/pgx/v5/pgxpool"

config, _ := pgxpool.ParseConfig("postgresql://...")
config.MaxConns = 20
config.MinConns = 5
config.MaxConnLifetime = time.Hour
config.MaxConnIdleTime = time.Minute * 10
pool, _ := pgxpool.NewWithConfig(context.Background(), config)

Serverless Considerations

Problem

Traditional connection pooling doesn't work well in serverless:

• Each function instance creates its own pool
• 100 concurrent functions × 10 connections = 1,000 database connections!
• Database runs out of connections (typically max 100-500)

Solution: pgBouncer

pgBouncer is a connection pooler that sits between your app and database.

Your App (1000 instances, 1 connection each) → pgBouncer (20 connections) → Database

Setup:

# Install pgBouncer
apt-get install pgbouncer

# Configure /etc/pgbouncer/pgbouncer.ini
[databases]
mydb = host=localhost port=5432 dbname=mydb

[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 20

Application Configuration:

# Connect to pgBouncer instead of PostgreSQL directly
engine = create_engine(
    "postgresql://user:pass@pgbouncer-host:6432/db",
    pool_size=1,  # Serverless: 1-2 connections per function
    max_overflow=0
)

Serverless Database Services

Modern serverless databases handle pooling automatically:

// Neon serverless driver (no pooling needed)
import { neon } from '@neondatabase/serverless'

const sql = neon(process.env.DATABASE_URL!)
const users = await sql`SELECT * FROM users`

Monitoring Connection Pools

PostgreSQL

-- Current connections
SELECT count(*) FROM pg_stat_activity;

-- Connections by state
SELECT state, count(*)
FROM pg_stat_activity
GROUP BY state;

-- Long-running connections
SELECT pid, now() - query_start AS duration, query
FROM pg_stat_activity
WHERE state = 'active' AND now() - query_start > interval '5 minutes';

-- Kill connection
SELECT pg_terminate_backend(pid);

Application Metrics

Track:

• Active connections
• Idle connections
• Connection wait time
• Connection errors

Python (SQLAlchemy):

# Check pool status
print(f"Pool size: {engine.pool.size()}")
print(f"Checked out: {engine.pool.checkedout()}")
print(f"Overflow: {engine.pool.overflow()}")

Rust (SQLx):

println!("Pool size: {}", pool.size());
println!("Idle connections: {}", pool.num_idle());

Best Practices

1. Right-Size Your Pool

Too few connections:

• Requests wait for connections
• High latency
• Poor throughput

Too many connections:

• Database CPU exhaustion
• Memory pressure
• Context switching overhead

Start with: pool_size = (num_cores * 2) + num_disks

2. Use Connection Lifecycle Management

# SQLAlchemy - verify connection before use
engine = create_engine(..., pool_pre_ping=True)

# Also recycle connections periodically
engine = create_engine(..., pool_recycle=3600)  # 1 hour

3. Handle Connection Errors

from sqlalchemy.exc import OperationalError
import time

def execute_with_retry(query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return session.execute(query)
        except OperationalError as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

4. Close Connections Properly

# Context manager (recommended)
with Session(engine) as session:
    user = session.query(User).first()

# Or manual close
session = Session(engine)
try:
    user = session.query(User).first()
finally:
    session.close()

5. Serverless: Use Connection Poolers

// Option 1: Use serverless-friendly database
const sql = neon(process.env.DATABASE_URL!)  // Neon handles pooling

// Option 2: Use pgBouncer
const pool = new Pool({
  connectionString: process.env.PGBOUNCER_URL,
  max: 1  // Serverless: 1 connection per function
})

// Option 3: Use Prisma Data Proxy (for Prisma users)
// DATABASE_URL="prisma://aws-us-east-1.prisma-data.com/?api_key=..."

Troubleshooting

"Too many connections" Error

Cause: Application pool size × number of instances > database max_connections

Solutions: 1. Reduce pool_size in application 2. Increase max_connections in PostgreSQL (requires more RAM) 3. Add pgBouncer connection pooler 4. Use serverless database (Neon, PlanetScale)

High Connection Wait Times

Cause: Pool too small for workload

Solutions: 1. Increase pool_size 2. Increase max_overflow for bursty traffic 3. Optimize slow queries (reduce connection hold time)

Stale Connections

Cause: Database closed connections but pool still holds them

Solutions: 1. Enable pool_pre_ping (SQLAlchemy) 2. Set pool_recycle to less than database timeout 3. Configure idle_timeout and max_lifetime (SQLx)

Resources

• pgBouncer: https://www.pgbouncer.org/
• PgPool II: https://www.pgpool.net/
• HikariCP (Java): https://github.com/brettwooldridge/HikariCP

Database Migrations Guide

Table of Contents

• Safe Migration Patterns
• Multi-Phase Deployments
• Common Patterns
• 1. Adding a Column
• 2. Dropping a Column
• 3. Renaming a Column
• 4. Creating Indexes
• 5. Adding Foreign Keys
• 6. Changing Column Type
• Migration Tools by Language
• Python (Alembic + SQLAlchemy)
• TypeScript (Prisma Migrate)
• TypeScript (Drizzle Kit)
• Rust (SQLx)
• Go (golang-migrate)
• Best Practices
• 1. Test Migrations in Staging
• 2. Measure Migration Duration
• 3. Concurrent Migrations
• 4. Monitor During Migration
• 5. Rollback Plan
• Common Pitfalls
• ❌ Don't Use Transactions for Long Operations
• ❌ Don't Assume Migration Speed
• ❌ Don't Mix Schema and Data Changes
• Resources

Safe Migration Patterns

Multi-Phase Deployments

Problem: Dropping columns or renaming tables can break running application instances during deployment.

Solution: Multi-phase migrations that maintain backward compatibility.

Common Patterns

1. Adding a Column

Safe (Single Phase):

-- Phase 1: Add column (nullable)
ALTER TABLE users ADD COLUMN phone VARCHAR(20);

-- Later: Make NOT NULL after backfill
-- ALTER TABLE users ALTER COLUMN phone SET NOT NULL;

Why Safe: New code can write to column, old code ignores it.

2. Dropping a Column

Unsafe:

-- DON'T DO THIS
ALTER TABLE users DROP COLUMN deprecated_field;
-- Running instances will crash!

Safe (Multi-Phase):

-- Phase 1: Make nullable, deploy code that doesn't use column
ALTER TABLE users ALTER COLUMN deprecated_field DROP NOT NULL;

-- Phase 2: Drop constraints, deploy
-- (Deploy and monitor)

-- Phase 3: Drop column after confirming no code uses it
-- ALTER TABLE users DROP COLUMN deprecated_field;

3. Renaming a Column

Unsafe:

-- DON'T DO THIS
ALTER TABLE users RENAME COLUMN email TO user_email;
-- Running instances will fail!

Safe (Multi-Phase):

-- Phase 1: Add new column
ALTER TABLE users ADD COLUMN user_email VARCHAR(255);

-- Phase 2: Backfill data, deploy code writing to BOTH columns
UPDATE users SET user_email = email WHERE user_email IS NULL;

-- Phase 3: Make NOT NULL, deploy code reading from new column
-- ALTER TABLE users ALTER COLUMN user_email SET NOT NULL;

-- Phase 4: Drop old column, deploy code using only new column
-- ALTER TABLE users DROP COLUMN email;

4. Creating Indexes

Unsafe (Blocks Writes):

-- DON'T DO THIS in production
CREATE INDEX idx_users_email ON users(email);
-- Locks table during creation!

Safe (PostgreSQL):

-- Create concurrently (doesn't block writes)
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);

Note: CONCURRENTLY cannot run inside a transaction. MySQL/MariaDB use ALGORITHM=INPLACE, LOCK=NONE.

5. Adding Foreign Keys

Safe Pattern:

-- Phase 1: Add column (nullable)
ALTER TABLE posts ADD COLUMN author_id INTEGER;

-- Phase 2: Backfill data
UPDATE posts SET author_id = (SELECT id FROM users WHERE users.email = posts.author_email);

-- Phase 3: Create index (important for performance!)
CREATE INDEX CONCURRENTLY idx_posts_author_id ON posts(author_id);

-- Phase 4: Add constraint
ALTER TABLE posts
ADD CONSTRAINT fk_posts_author
FOREIGN KEY (author_id)
REFERENCES users(id)
ON DELETE CASCADE;

-- Phase 5: Make NOT NULL
-- ALTER TABLE posts ALTER COLUMN author_id SET NOT NULL;

6. Changing Column Type

Unsafe:

-- DON'T DO THIS
ALTER TABLE users ALTER COLUMN age TYPE BIGINT;
-- Rewrites entire table, locks for duration!

Safe (Multi-Phase):

-- Phase 1: Add new column with new type
ALTER TABLE users ADD COLUMN age_bigint BIGINT;

-- Phase 2: Backfill, deploy code writing to both
UPDATE users SET age_bigint = age WHERE age_bigint IS NULL;

-- Phase 3: Drop old column, deploy code using new column
-- ALTER TABLE users DROP COLUMN age;
-- ALTER TABLE users RENAME COLUMN age_bigint TO age;

Migration Tools by Language

Python (Alembic + SQLAlchemy)

# Initialize
alembic init alembic

# Create migration
alembic revision --autogenerate -m "Add users table"

# Apply
alembic upgrade head

# Rollback
alembic downgrade -1

Example Migration:

# alembic/versions/001_add_users.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email', 'users')
    op.drop_table('users')

TypeScript (Prisma Migrate)

# Create migration
npx prisma migrate dev --name add_users

# Apply to production
npx prisma migrate deploy

# Reset (development only!)
npx prisma migrate reset

TypeScript (Drizzle Kit)

# Generate migration
npx drizzle-kit generate:pg

# Apply migration
npx drizzle-kit push:pg

# Or use custom migration runner

Rust (SQLx)

# Create migration
sqlx migrate add add_users

# Run migrations
sqlx migrate run

# Revert
sqlx migrate revert

Embed in Binary:

sqlx::migrate!("./migrations")
    .run(&pool)
    .await?;

Go (golang-migrate)

# Create migration
migrate create -ext sql -dir migrations -seq add_users

# Run migrations
migrate -path migrations -database "postgresql://..." up

# Rollback
migrate -path migrations -database "postgresql://..." down 1

Best Practices

1. Test Migrations in Staging

# Copy production data to staging
pg_dump -h prod -d mydb | psql -h staging -d mydb_staging

# Run migration in staging
alembic upgrade head

# Test application against staging

# If successful, run in production

2. Measure Migration Duration

# Time migration
time alembic upgrade head

# For large tables (>1M rows), expect:
# - Adding column: Seconds
# - Creating index: Minutes to hours
# - Changing column type: Minutes to hours (table rewrite)

3. Concurrent Migrations

PostgreSQL Specific:

-- Create index without blocking writes
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);

-- If it fails, drop and retry (can't retry CONCURRENTLY in transaction)
DROP INDEX CONCURRENTLY IF EXISTS idx_users_email;
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);

-- Add constraint without long lock (PostgreSQL 12+)
ALTER TABLE posts ADD CONSTRAINT fk_posts_author
FOREIGN KEY (author_id) REFERENCES users(id) NOT VALID;

-- Validate separately (acquires brief lock)
ALTER TABLE posts VALIDATE CONSTRAINT fk_posts_author;

4. Monitor During Migration

-- Check progress (PostgreSQL)
SELECT pid, now() - query_start AS duration, state, query
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY duration DESC;

-- Check locks
SELECT blocked_locks.pid AS blocked_pid,
       blocking_locks.pid AS blocking_pid,
       blocked_activity.query AS blocked_query,
       blocking_activity.query AS current_query
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks ON blocking_locks.locktype = blocked_locks.locktype
JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

5. Rollback Plan

Always have rollback:

-- Migration (up)
CREATE TABLE new_table (...);

-- Rollback (down)
DROP TABLE new_table;

Test rollback in staging:

alembic upgrade head  # Apply
alembic downgrade -1  # Rollback
alembic upgrade head  # Re-apply

Common Pitfalls

❌ Don't Use Transactions for Long Operations

-- BAD - Locks for entire duration
BEGIN;
CREATE INDEX idx_users_email ON users(email);  -- Takes 10 minutes
COMMIT;

-- GOOD - No transaction, use CONCURRENTLY
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);

❌ Don't Assume Migration Speed

# BAD - No timeout
def upgrade():
    op.execute("UPDATE large_table SET column = ...")  # Could take hours!

# GOOD - Add timeout or break into batches
def upgrade():
    conn = op.get_bind()
    conn.execute(text("SET statement_timeout = '10min'"))
    # Or batch updates:
    # UPDATE large_table SET column = ... WHERE id BETWEEN 1 AND 10000;

❌ Don't Mix Schema and Data Changes

# BAD
def upgrade():
    op.create_table('users', ...)
    op.execute("INSERT INTO users VALUES (...)")  # Data migration

# GOOD - Separate migrations
# 001_create_users_table.py - Schema
# 002_populate_initial_users.py - Data

Resources

• Alembic: https://alembic.sqlalchemy.org/
• Prisma Migrate: https://www.prisma.io/docs/concepts/components/prisma-migrate
• Drizzle Kit: https://orm.drizzle.team/kit-docs/overview
• SQLx Migrations: https://github.com/launchbadge/sqlx/tree/main/sqlx-cli
• golang-migrate: https://github.com/golang-migrate/migrate

MySQL Implementation Guide

MySQL-specific patterns, PlanetScale integration, and best practices for production deployments.

Table of Contents

• When to Choose MySQL
• Installation
• Connection Configuration
• Python (SQLAlchemy)
• TypeScript (Prisma)
• Go (database/sql)
• Schema Design Differences
• JSON Support
• Auto-Increment Primary Keys
• UUID Primary Keys
• PlanetScale Integration
• Non-Blocking Schema Changes
• Prisma + PlanetScale Setup
• Database Branching
• Full-Text Search
• Replication Setup
• Primary-Replica Replication
• Performance Optimization
• Index Strategy
• Query Optimization
• Connection Pooling
• MySQL vs PostgreSQL Syntax Differences
• Backup and Restore
• Logical Backup (mysqldump)
• Point-in-Time Recovery
• Migration from MySQL to PostgreSQL
• Serverless: PlanetScale vs Neon (PostgreSQL)
• Best Practices
• Common Pitfalls
• Resources

When to Choose MySQL

• Legacy system compatibility - Existing MySQL infrastructure
• PlanetScale serverless - Non-blocking schema changes, branch-based workflows
• WordPress/PHP ecosystems - Native integration
• Read replicas - Built-in replication

When PostgreSQL is better: JSON operations, advanced indexing (GiST, GIN), extensions (PostGIS, pgvector)

Installation

# Docker
docker run --name mysql -e MYSQL_ROOT_PASSWORD=password -p 3306:3306 -d mysql:8.2

# Ubuntu/Debian
sudo apt update
sudo apt install mysql-server

# macOS (Homebrew)
brew install mysql
brew services start mysql

Connection Configuration

Python (SQLAlchemy)

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "mysql+pymysql://user:password@localhost:3306/dbname?charset=utf8mb4",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # Verify connections before use
    pool_recycle=3600,   # Recycle connections every hour
    echo=False,
)

TypeScript (Prisma)

// schema.prisma
datasource db {
  provider = "mysql"
  url      = env("DATABASE_URL")
  relationMode = "prisma"  // For PlanetScale (no foreign keys)
}

// .env
DATABASE_URL="mysql://user:password@localhost:3306/dbname"

Go (database/sql)

import (
    "database/sql"
    _ "github.com/go-sql-driver/mysql"
)

dsn := "user:password@tcp(localhost:3306)/dbname?parseTime=true&charset=utf8mb4"
db, err := sql.Open("mysql", dsn)
db.SetMaxOpenConns(25)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(time.Hour)

Schema Design Differences

JSON Support

MySQL 8.0+:

CREATE TABLE products (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(255) NOT NULL,
  attributes JSON,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Query JSON
SELECT * FROM products WHERE JSON_EXTRACT(attributes, '$.color') = 'red';

-- Index JSON path
ALTER TABLE products ADD COLUMN color VARCHAR(50)
  GENERATED ALWAYS AS (JSON_UNQUOTE(JSON_EXTRACT(attributes, '$.color'))) STORED;
CREATE INDEX idx_color ON products(color);

Auto-Increment Primary Keys

MySQL default (vs PostgreSQL SERIAL):

CREATE TABLE users (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  email VARCHAR(255) UNIQUE NOT NULL,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

UUID Primary Keys

-- MySQL 8.0+ has UUID() function
CREATE TABLE sessions (
  id BINARY(16) PRIMARY KEY DEFAULT (UUID_TO_BIN(UUID())),
  user_id BIGINT NOT NULL,
  expires_at TIMESTAMP NOT NULL
);

-- Query with UUID
SELECT * FROM sessions WHERE id = UUID_TO_BIN('550e8400-e29b-41d4-a716-446655440000');

PlanetScale Integration

Non-Blocking Schema Changes

PlanetScale's killer feature: Schema changes don't lock tables.

Deploy request workflow:

# 1. Create development branch
pscale branch create mydb dev-feature

# 2. Make schema changes
pscale shell mydb dev-feature
mysql> ALTER TABLE users ADD COLUMN phone VARCHAR(20);

# 3. Create deploy request
pscale deploy-request create mydb dev-feature

# 4. Review and deploy (zero downtime)
pscale deploy-request deploy mydb 1

# 5. Merge to main
# PlanetScale automatically merges after successful deploy

Prisma + PlanetScale Setup

// schema.prisma
datasource db {
  provider     = "mysql"
  url          = env("DATABASE_URL")
  relationMode = "prisma"  // Required for PlanetScale (no FK constraints)
}

generator client {
  provider = "prisma-client-js"
}

model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  posts     Post[]
}

model Post {
  id       Int    @id @default(autoincrement())
  userId   Int
  title    String
  user     User   @relation(fields: [userId], references: [id])

  @@index([userId])  // Manual index (no FK constraint)
}

Connection string:

# PlanetScale connection (from dashboard)
DATABASE_URL="mysql://username:password@aws.connect.psdb.cloud/dbname?ssl={"rejectUnauthorized":true}"

Database Branching

# Create branch from main
pscale branch create mydb feature-xyz --from main

# Connect to branch
pscale connect mydb feature-xyz --port 3309

# Run migrations on branch
npm run prisma migrate dev

# Deploy to main via deploy request
pscale deploy-request create mydb feature-xyz

Full-Text Search

CREATE TABLE articles (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  title VARCHAR(255) NOT NULL,
  content TEXT NOT NULL,
  FULLTEXT(title, content)
);

-- Full-text search
SELECT * FROM articles
WHERE MATCH(title, content) AGAINST('database optimization' IN NATURAL LANGUAGE MODE);

-- Boolean mode (AND, OR, NOT)
SELECT * FROM articles
WHERE MATCH(title, content) AGAINST('+MySQL -PostgreSQL' IN BOOLEAN MODE);

Replication Setup

Primary-Replica Replication

Primary server (my.cnf):

[mysqld]
server-id = 1
log_bin = /var/log/mysql/mysql-bin.log
binlog_do_db = mydb

Replica server (my.cnf):

[mysqld]
server-id = 2
relay-log = /var/log/mysql/mysql-relay-bin
log_bin = /var/log/mysql/mysql-bin.log
read_only = 1

Configure replication:

-- On primary
CREATE USER 'replicator'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'replicator'@'%';
SHOW MASTER STATUS;  -- Note File and Position

-- On replica
CHANGE MASTER TO
  MASTER_HOST='primary-host',
  MASTER_USER='replicator',
  MASTER_PASSWORD='password',
  MASTER_LOG_FILE='mysql-bin.000001',
  MASTER_LOG_POS=107;

START SLAVE;
SHOW SLAVE STATUS\G

Performance Optimization

Index Strategy

Composite index order matters:

-- Query: WHERE user_id = ? AND created_at > ? ORDER BY created_at DESC
CREATE INDEX idx_user_created ON posts(user_id, created_at DESC);

-- Query: WHERE status = ? AND priority = ? ORDER BY created_at DESC
CREATE INDEX idx_status_priority_created ON tasks(status, priority, created_at DESC);

Covering indexes:

-- Query: SELECT id, title, created_at FROM posts WHERE user_id = ?
CREATE INDEX idx_user_id_covering ON posts(user_id, id, title, created_at);

Query Optimization

Use EXPLAIN to analyze queries:

EXPLAIN SELECT * FROM users WHERE email = 'test@example.com';

EXPLAIN FORMAT=JSON SELECT u.*, COUNT(p.id) as post_count
FROM users u
LEFT JOIN posts p ON u.id = p.user_id
GROUP BY u.id;

Connection Pooling

Optimal pool sizes:

# Formula: (core_count * 2) + effective_spindle_count
# For 4 cores + SSD: (4 * 2) + 1 = 9 ≈ 10

engine = create_engine(
    "mysql+pymysql://...",
    pool_size=10,
    max_overflow=20,
)

MySQL vs PostgreSQL Syntax Differences

Backup and Restore

Logical Backup (mysqldump)

# Full database backup
mysqldump -u root -p --single-transaction --routines --triggers mydb > backup.sql

# Restore
mysql -u root -p mydb < backup.sql

# Compressed backup
mysqldump -u root -p mydb | gzip > backup.sql.gz

# Restore from compressed
gunzip < backup.sql.gz | mysql -u root -p mydb

Point-in-Time Recovery

# Enable binary logging (my.cnf)
[mysqld]
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW
expire_logs_days = 7

# Restore to specific point in time
mysqlbinlog --stop-datetime="2025-12-03 10:30:00" mysql-bin.000001 | mysql -u root -p mydb

Migration from MySQL to PostgreSQL

Use pgLoader for automated migration:

pgloader mysql://user:pass@localhost/mydb postgresql://user:pass@localhost/mydb

# Custom migration script
pgloader mysql.load

# mysql.load
LOAD DATABASE
  FROM mysql://root@localhost/mydb
  INTO postgresql://user@localhost/mydb
  WITH include drop, create tables, create indexes
  SET maintenance_work_mem to '512MB',
      work_mem to '128MB';

Serverless: PlanetScale vs Neon (PostgreSQL)

Choose PlanetScale if:

• Non-blocking schema changes are critical
• MySQL ecosystem required
• Read replicas needed

Choose Neon if:

• Need PostgreSQL features (JSONB, arrays, extensions)
• Want scale-to-zero compute
• Database branching for preview environments

Best Practices

1. Use utf8mb4 (not utf8) for full Unicode support 2. Enable strict SQL mode - Prevents silent data truncation 3. Use InnoDB engine (default in MySQL 8.0+) 4. Index foreign keys manually (especially with PlanetScale) 5. Use prepared statements to prevent SQL injection 6. Monitor slow query log regularly 7. Set appropriate connection pool sizes (10-20 for web APIs) 8. *Use `SELECT FROM sparingly** (specify columns) 9. **Avoid OR in WHERE clauses** (use UNION` instead) 10. Backup before schema changes (even with PlanetScale)

Common Pitfalls

Implicit type conversion:

-- ❌ Slow (varchar column, int comparison)
SELECT * FROM users WHERE id = 123;  -- id is VARCHAR

-- ✅ Fast (proper type)
SELECT * FROM users WHERE id = '123';

N+1 queries:

# ❌ N+1 problem
users = db.query(User).all()
for user in users:
    posts = db.query(Post).filter(Post.user_id == user.id).all()  # N queries

# ✅ Eager loading
users = db.query(User).options(joinedload(User.posts)).all()  # 1 query

Resources

• MySQL Docs: https://dev.mysql.com/doc/
• PlanetScale Docs: https://planetscale.com/docs
• MySQL Performance Blog: https://www.percona.com/blog/

Go Database Libraries

Table of Contents

• Overview
• sqlc (Recommended)
• Installation
• Configuration
• Schema
• Queries
• Generate Code
• Usage
• Gin Integration
• GORM (Rapid Development)
• Installation
• Usage
• pgx (PostgreSQL Driver)
• Installation
• Usage
• Comparison
• Choose sqlc When:
• Choose GORM When:
• Choose pgx When:
• Best Practices
• Connection Pooling
• Error Handling
• Migrations
• Resources

Overview

sqlc (Recommended)

Key Feature: Generates type-safe Go code from SQL queries.

Installation

go install github.com/sqlc-dev/sqlc/cmd/sqlc@latest

Configuration

# sqlc.yaml
version: "2"
sql:
  - engine: "postgresql"
    queries: "queries.sql"
    schema: "schema.sql"
    gen:
      go:
        package: "db"
        out: "db"
        emit_json_tags: true
        emit_prepared_queries: false
        emit_interface: false
        emit_exact_table_names: false

Schema

-- schema.sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_users_email ON users(email);

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT,
    published BOOLEAN DEFAULT false,
    author_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Queries

-- queries.sql

-- name: GetUser :one
SELECT id, email, name, created_at FROM users WHERE id = $1;

-- name: GetUserByEmail :one
SELECT id, email, name, created_at FROM users WHERE email = $1;

-- name: ListUsers :many
SELECT id, email, name, created_at FROM users ORDER BY id;

-- name: CreateUser :one
INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name, created_at;

-- name: UpdateUser :exec
UPDATE users SET name = $1 WHERE id = $2;

-- name: DeleteUser :exec
DELETE FROM users WHERE id = $1;

-- name: ListUserPosts :many
SELECT p.id, p.title, p.content, p.published
FROM posts p
JOIN users u ON p.author_id = u.id
WHERE u.id = $1
ORDER BY p.created_at DESC;

Generate Code

sqlc generate

Usage

package main

import (
    "context"
    "database/sql"
    "log"
    "yourproject/db"

    _ "github.com/lib/pq"
)

func main() {
    // Connect
    conn, err := sql.Open("postgres", "postgresql://user:pass@localhost/db")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // Connection pooling
    conn.SetMaxOpenConns(20)
    conn.SetMaxIdleConns(10)
    conn.SetConnMaxLifetime(time.Hour)

    // Create queries instance
    queries := db.New(conn)
    ctx := context.Background()

    // Create
    user, err := queries.CreateUser(ctx, db.CreateUserParams{
        Email: "test@example.com",
        Name:  "Test User",
    })
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Created user: %+v\n", user)

    // Read
    user, err = queries.GetUser(ctx, user.ID)
    if err != nil {
        log.Fatal(err)
    }

    users, err := queries.ListUsers(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // Update
    err = queries.UpdateUser(ctx, db.UpdateUserParams{
        ID:   user.ID,
        Name: "Updated Name",
    })
    if err != nil {
        log.Fatal(err)
    }

    // Delete
    err = queries.DeleteUser(ctx, user.ID)
    if err != nil {
        log.Fatal(err)
    }

    // Transactions
    tx, err := conn.Begin()
    if err != nil {
        log.Fatal(err)
    }
    defer tx.Rollback()

    qtx := queries.WithTx(tx)
    _, err = qtx.CreateUser(ctx, db.CreateUserParams{
        Email: "user1@example.com",
        Name:  "User 1",
    })
    if err != nil {
        log.Fatal(err)
    }

    _, err = qtx.CreateUser(ctx, db.CreateUserParams{
        Email: "user2@example.com",
        Name:  "User 2",
    })
    if err != nil {
        log.Fatal(err)
    }

    if err = tx.Commit(); err != nil {
        log.Fatal(err)
    }
}

Gin Integration

package main

import (
    "database/sql"
    "net/http"
    "yourproject/db"

    "github.com/gin-gonic/gin"
    _ "github.com/lib/pq"
)

func main() {
    conn, _ := sql.Open("postgres", "postgresql://...")
    defer conn.Close()

    queries := db.New(conn)

    r := gin.Default()

    r.POST("/users", func(c *gin.Context) {
        var req struct {
            Email string `json:"email" binding:"required"`
            Name  string `json:"name" binding:"required"`
        }

        if err := c.ShouldBindJSON(&req); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
            return
        }

        user, err := queries.CreateUser(c.Request.Context(), db.CreateUserParams{
            Email: req.Email,
            Name:  req.Name,
        })
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
            return
        }

        c.JSON(http.StatusCreated, user)
    })

    r.GET("/users/:id", func(c *gin.Context) {
        var uri struct {
            ID int32 `uri:"id" binding:"required"`
        }

        if err := c.ShouldBindUri(&uri); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
            return
        }

        user, err := queries.GetUser(c.Request.Context(), uri.ID)
        if err != nil {
            c.JSON(http.StatusNotFound, gin.H{"error": "User not found"})
            return
        }

        c.JSON(http.StatusOK, user)
    })

    r.Run(":8080")
}

Why sqlc?

• Type-safe generated code
• SQL-first (full SQL control)
• Zero runtime overhead
• Easy to understand (just Go code)
• Excellent for PostgreSQL

GORM (Rapid Development)

Installation

go get -u gorm.io/gorm
go get -u gorm.io/driver/postgres

Usage

package main

import (
    "gorm.io/driver/postgres"
    "gorm.io/gorm"
    "time"
)

type User struct {
    ID        uint      `gorm:"primaryKey" json:"id"`
    Email     string    `gorm:"uniqueIndex;not null" json:"email"`
    Name      string    `gorm:"not null" json:"name"`
    CreatedAt time.Time `json:"created_at"`
    Posts     []Post    `gorm:"foreignKey:AuthorID" json:"posts,omitempty"`
}

type Post struct {
    ID        uint      `gorm:"primaryKey" json:"id"`
    Title     string    `gorm:"not null" json:"title"`
    Content   string    `json:"content"`
    Published bool      `gorm:"default:false" json:"published"`
    AuthorID  uint      `gorm:"not null" json:"author_id"`
    CreatedAt time.Time `json:"created_at"`
}

func main() {
    dsn := "host=localhost user=user password=pass dbname=db port=5432 sslmode=disable"
    db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
    if err != nil {
        panic(err)
    }

    // Auto-migrate (creates tables)
    db.AutoMigrate(&User{}, &Post{})

    // Create
    user := User{Email: "test@example.com", Name: "Test User"}
    db.Create(&user)

    // Read
    var foundUser User
    db.First(&foundUser, "email = ?", "test@example.com")

    var users []User
    db.Where("email LIKE ?", "%@example.com").Find(&users)

    // Update
    db.Model(&foundUser).Update("name", "Updated Name")

    // Delete
    db.Delete(&foundUser)

    // Preload relations
    var userWithPosts User
    db.Preload("Posts").First(&userWithPosts, user.ID)

    // Transactions
    db.Transaction(func(tx *gorm.DB) error {
        if err := tx.Create(&User{Email: "user1@example.com", Name: "User 1"}).Error; err != nil {
            return err
        }
        if err := tx.Create(&User{Email: "user2@example.com", Name: "User 2"}).Error; err != nil {
            return err
        }
        return nil
    })
}

Why GORM?

• Rapid development
• Active Record pattern
• Auto-migrations
• Association handling
• Large community

pgx (PostgreSQL Driver)

High-performance PostgreSQL driver.

Installation

go get github.com/jackc/pgx/v5
go get github.com/jackc/pgx/v5/pgxpool

Usage

package main

import (
    "context"
    "github.com/jackc/pgx/v5/pgxpool"
    "log"
)

type User struct {
    ID    int32
    Email string
    Name  string
}

func main() {
    ctx := context.Background()

    // Connection pool
    pool, err := pgxpool.New(ctx, "postgresql://user:pass@localhost/db")
    if err != nil {
        log.Fatal(err)
    }
    defer pool.Close()

    // Query
    var user User
    err = pool.QueryRow(ctx, "SELECT id, email, name FROM users WHERE id = $1", 1).
        Scan(&user.ID, &user.Email, &user.Name)

    // Insert
    err = pool.QueryRow(ctx,
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id",
        "test@example.com", "Test User").Scan(&user.ID)

    // Transaction
    tx, err := pool.Begin(ctx)
    if err != nil {
        log.Fatal(err)
    }
    defer tx.Rollback(ctx)

    _, err = tx.Exec(ctx, "INSERT INTO users (email, name) VALUES ($1, $2)", "user1@example.com", "User 1")
    if err != nil {
        log.Fatal(err)
    }

    err = tx.Commit(ctx)
}

Why pgx?

• Best PostgreSQL performance
• Low-level control
• Connection pooling built-in
• PostgreSQL-specific features

Comparison

Choose sqlc When:

• Want type safety from SQL
• Prefer SQL-first approach
• Need generated code
• PostgreSQL/MySQL/SQLite

Choose GORM When:

• Rapid development
• Like Active Record
• Need auto-migrations
• Want associations

Choose pgx When:

• PostgreSQL-only
• Maximum performance
• Low-level control
• Simple queries

Best Practices

Connection Pooling

// database/sql
db.SetMaxOpenConns(20)
db.SetMaxIdleConns(10)
db.SetConnMaxLifetime(time.Hour)
db.SetConnMaxIdleTime(time.Minute * 10)

// pgxpool
config, _ := pgxpool.ParseConfig("postgresql://...")
config.MaxConns = 20
config.MinConns = 5
config.MaxConnLifetime = time.Hour
config.MaxConnIdleTime = time.Minute * 10
pool, _ := pgxpool.NewWithConfig(context.Background(), config)

Error Handling

import "errors"

var (
    ErrNotFound      = errors.New("not found")
    ErrDuplicateEmail = errors.New("duplicate email")
)

func GetUser(ctx context.Context, queries *db.Queries, id int32) (db.User, error) {
    user, err := queries.GetUser(ctx, id)
    if err != nil {
        if errors.Is(err, sql.ErrNoRows) {
            return db.User{}, ErrNotFound
        }
        return db.User{}, fmt.Errorf("get user: %w", err)
    }
    return user, nil
}

Migrations

# Install golang-migrate
go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest

# Create migration
migrate create -ext sql -dir migrations -seq add_users

# Run migrations
migrate -path migrations -database "postgresql://..." up

-- migrations/000001_add_users.up.sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL
);

-- migrations/000001_add_users.down.sql
DROP TABLE users;

Resources

• sqlc: https://sqlc.dev/
• GORM: https://gorm.io/
• pgx: https://github.com/jackc/pgx
• golang-migrate: https://github.com/golang-migrate/migrate

Python ORMs and Database Libraries

Table of Contents

• Overview
• SQLAlchemy 2.0 (Recommended for Most Projects)
• Core Concepts
• ORM Pattern (Declarative)
• Async Support (SQLAlchemy 2.0+)
• Core (Query Builder)
• Transactions
• SQLModel (FastAPI Integration)
• asyncpg (High Performance)
• Tortoise ORM (Async-First)
• Comparison: When to Use Each
• Choose SQLAlchemy 2.0 When:
• Choose SQLModel When:
• Choose Tortoise ORM When:
• Choose asyncpg When:
• Migration Tools
• Alembic (SQLAlchemy)
• Aerich (Tortoise ORM)
• Best Practices
• Use Connection Pooling
• Use Transactions for Multi-Statement Operations
• Use Parameterized Queries (Prevent SQL Injection)
• Batch Operations for Performance
• Index Frequently Queried Columns
• Resources

Overview

Python database libraries span from high-level ORMs to low-level drivers, each with different trade-offs.

SQLAlchemy 2.0 (Recommended for Most Projects)

Context7: /websites/sqlalchemy_en_21 (7,090 snippets) ORM Docs: /websites/sqlalchemy_en_20_orm (2,047 snippets)

Core Concepts

SQLAlchemy 2.0 has two layers: 1. Core: Low-level SQL expression language (query builder) 2. ORM: High-level object-relational mapping

Installation:

pip install sqlalchemy[asyncio] asyncpg  # For PostgreSQL
pip install sqlalchemy[asyncio] aiomysql  # For MySQL

ORM Pattern (Declarative)

Define Models:

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False, index=True)
    name = Column(String, nullable=False)
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    author = relationship("User", back_populates="posts")

# Create engine with connection pooling
engine = create_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True
)

# Create tables
Base.metadata.create_all(engine)

CRUD Operations:

# Create
with Session(engine) as session:
    user = User(email="test@example.com", name="Test User")
    session.add(user)
    session.commit()
    session.refresh(user)  # Get generated ID
    print(user.id)

# Read
with Session(engine) as session:
    user = session.query(User).filter(User.email == "test@example.com").first()
    # Or using modern select() syntax
    from sqlalchemy import select
    stmt = select(User).where(User.email == "test@example.com")
    user = session.execute(stmt).scalar_one()

# Update
with Session(engine) as session:
    user = session.query(User).filter(User.id == 1).first()
    user.name = "Updated Name"
    session.commit()

# Delete
with Session(engine) as session:
    user = session.query(User).filter(User.id == 1).first()
    session.delete(user)
    session.commit()

Relationships and Joins:

with Session(engine) as session:
    # Eager loading (fetch posts with users in one query)
    from sqlalchemy.orm import joinedload
    users = session.query(User).options(joinedload(User.posts)).all()

    # Explicit join
    stmt = (
        select(User, Post)
        .join(Post)
        .where(User.email.like('%@example.com'))
    )
    results = session.execute(stmt).all()

Async Support (SQLAlchemy 2.0+)

Async Engine:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10
)

async_session = async_sessionmaker(async_engine, expire_on_commit=False)

# Usage with FastAPI
from fastapi import Depends

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()
    return user

Core (Query Builder)

Use Core for complex queries, better performance, or when ORM overhead unnecessary.

from sqlalchemy import Table, Column, Integer, String, MetaData, select, insert

metadata = MetaData()

users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('email', String, unique=True, nullable=False),
    Column('name', String, nullable=False)
)

# Create
stmt = insert(users).values(email="test@example.com", name="Test User")
with engine.connect() as conn:
    result = conn.execute(stmt)
    conn.commit()

# Read
stmt = select(users).where(users.c.email == "test@example.com")
with engine.connect() as conn:
    result = conn.execute(stmt)
    row = result.fetchone()

# Complex query
from sqlalchemy import func
stmt = (
    select(users.c.name, func.count(posts.c.id).label('post_count'))
    .select_from(users.join(posts))
    .group_by(users.c.name)
    .having(func.count(posts.c.id) > 5)
)

Transactions

from sqlalchemy import text

with Session(engine) as session:
    try:
        session.begin()  # Explicit transaction
        session.execute(text("UPDATE users SET name = 'Alice' WHERE id = 1"))
        session.execute(text("UPDATE posts SET title = 'New Title' WHERE user_id = 1"))
        session.commit()
    except Exception:
        session.rollback()
        raise

SQLModel (FastAPI Integration)

Installation:

pip install sqlmodel

Define Models:

from sqlmodel import SQLModel, Field, create_engine, Session, Relationship
from typing import Optional, List

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    name: str
    posts: List["Post"] = Relationship(back_populates="author")

class Post(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: Optional[str] = None
    user_id: int = Field(foreign_key="user.id")
    author: User = Relationship(back_populates="posts")

engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)

FastAPI Integration:

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select

app = FastAPI()

def get_session():
    with Session(engine) as session:
        yield session

@app.post("/users/", response_model=User)
def create_user(user: User, session: Session = Depends(get_session)):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/", response_model=List[User])
def list_users(skip: int = 0, limit: int = 100, session: Session = Depends(get_session)):
    statement = select(User).offset(skip).limit(limit)
    users = session.exec(statement).all()
    return users

Why SQLModel?

• Pydantic validation built-in
• Single model class for DB and API
• Type hints throughout
• FastAPI integration seamless
• Built on SQLAlchemy (can drop to SQLAlchemy for complex queries)

asyncpg (High Performance)

Direct PostgreSQL driver with best performance.

Installation:

pip install asyncpg

Usage:

import asyncpg

async def main():
    # Create connection pool
    pool = await asyncpg.create_pool(
        user='user',
        password='password',
        database='database',
        host='localhost',
        min_size=5,
        max_size=20
    )

    # Query
    async with pool.acquire() as conn:
        # Select
        rows = await conn.fetch('SELECT * FROM users WHERE email = $1', 'test@example.com')
        for row in rows:
            print(row['id'], row['email'], row['name'])

        # Insert
        user_id = await conn.fetchval(
            'INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id',
            'new@example.com', 'New User'
        )

        # Transaction
        async with conn.transaction():
            await conn.execute('UPDATE users SET name = $1 WHERE id = $2', 'Alice', 1)
            await conn.execute('UPDATE posts SET title = $1 WHERE user_id = $2', 'New Title', 1)

    await pool.close()

When to Use asyncpg:

• Maximum performance needed
• Willing to write raw SQL
• PostgreSQL-specific features
• Microservices with simple queries

Tortoise ORM (Async-First)

Installation:

pip install tortoise-orm asyncpg  # For PostgreSQL

Define Models:

from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator

class User(models.Model):
    id = fields.IntField(pk=True)
    email = fields.CharField(max_length=255, unique=True, index=True)
    name = fields.CharField(max_length=255)
    posts = fields.ReverseRelation["Post"]

    class Meta:
        table = "users"

class Post(models.Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=255)
    content = fields.TextField(null=True)
    author = fields.ForeignKeyField('models.User', related_name='posts')

    class Meta:
        table = "posts"

# Generate Pydantic models
User_Pydantic = pydantic_model_creator(User, name="User")
UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", exclude_readonly=True)

FastAPI Integration:

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

@app.post("/users/", response_model=User_Pydantic)
async def create_user(user: UserIn_Pydantic):
    user_obj = await User.create(**user.dict())
    return await User_Pydantic.from_tortoise_orm(user_obj)

@app.get("/users/{user_id}", response_model=User_Pydantic)
async def get_user(user_id: int):
    return await User_Pydantic.from_queryset_single(User.get(id=user_id))

register_tortoise(
    app,
    db_url="postgres://user:pass@localhost:5432/db",
    modules={"models": ["myapp.models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

Comparison: When to Use Each

Choose SQLAlchemy 2.0 When:

• Need maximum flexibility (ORM + Core)
• Working with complex schemas
• Require advanced features (polymorphic inheritance, hybrid properties)
• Want to switch between sync/async
• Team experienced with SQLAlchemy

Choose SQLModel When:

• Using FastAPI
• Want Pydantic integration
• Prefer simple, type-safe API
• Rapid prototyping
• Can drop to SQLAlchemy for complex queries

Choose Tortoise ORM When:

• Async-first application
• Like Django ORM syntax
• Need Pydantic models from ORM
• Simpler learning curve than SQLAlchemy

Choose asyncpg When:

• Maximum performance critical
• Simple queries, minimal ORM needed
• PostgreSQL-only application
• Microservices architecture

Migration Tools

Alembic (SQLAlchemy)

Installation:

pip install alembic
alembic init alembic

Create Migration:

alembic revision --autogenerate -m "Add users table"
alembic upgrade head  # Apply migration
alembic downgrade -1  # Rollback one migration

Migration File:

# alembic/versions/001_add_users.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(), nullable=False),
        sa.Column('name', sa.String(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email', 'users')
    op.drop_table('users')

Aerich (Tortoise ORM)

Installation:

pip install aerich
aerich init -t config.TORTOISE_ORM
aerich init-db

Create Migration:

aerich migrate --name "add_users"
aerich upgrade
aerich downgrade

Best Practices

Use Connection Pooling

# SQLAlchemy
engine = create_engine(
    "postgresql+asyncpg://...",
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=3600,
    pool_pre_ping=True
)

# asyncpg
pool = await asyncpg.create_pool(
    ...,
    min_size=10,
    max_size=20,
    max_queries=50000,  # Recycle connection after 50k queries
    max_inactive_connection_lifetime=300  # Recycle after 5 min idle
)

Use Transactions for Multi-Statement Operations

# SQLAlchemy
with Session(engine) as session:
    try:
        session.begin()
        session.add(user)
        session.add(post)
        session.commit()
    except Exception:
        session.rollback()
        raise

# asyncpg
async with conn.transaction():
    await conn.execute(...)
    await conn.execute(...)

Use Parameterized Queries (Prevent SQL Injection)

# GOOD - Parameterized
user = session.query(User).filter(User.email == email).first()
stmt = select(User).where(User.email == email)

# BAD - String interpolation (SQL injection risk!)
session.execute(f"SELECT * FROM users WHERE email = '{email}'")

Batch Operations for Performance

# SQLAlchemy bulk insert
session.bulk_insert_mappings(User, [
    {"email": "user1@example.com", "name": "User 1"},
    {"email": "user2@example.com", "name": "User 2"},
    # ... thousands of rows
])

# asyncpg batch insert
await conn.executemany(
    'INSERT INTO users (email, name) VALUES ($1, $2)',
    [('user1@example.com', 'User 1'), ('user2@example.com', 'User 2')]
)

Index Frequently Queried Columns

class User(Base):
    __tablename__ = 'users'
    email = Column(String, unique=True, index=True)  # Index for WHERE clauses
    created_at = Column(DateTime, index=True)  # Index for ORDER BY

Resources

• SQLAlchemy 2.0 Docs: https://docs.sqlalchemy.org/en/20/
• SQLModel Docs: https://sqlmodel.tiangolo.com/
• Tortoise ORM Docs: https://tortoise.github.io/
• asyncpg Docs: https://magicstack.github.io/asyncpg/
• Alembic Docs: https://alembic.sqlalchemy.org/

Rust Database Libraries

Table of Contents

• Overview
• SQLx (Recommended)
• Installation
• Setup
• Migrations
• Usage
• Compile-Time Checked Macros
• Axum Integration
• SeaORM (Full ORM)
• Installation
• Entity Definition
• Usage
• Diesel (Mature, Stable)
• Installation
• Schema Definition
• Usage
• Comparison
• Choose SQLx When:
• Choose SeaORM When:
• Choose Diesel When:
• Best Practices
• Connection Pooling
• Error Handling
• Transactions
• Resources

Overview

SQLx (Recommended)

Key Feature: Compile-time query validation - queries are checked against your database schema at build time.

Installation

# Cargo.toml
[dependencies]
sqlx = { version = "0.8", features = ["postgres", "runtime-tokio", "tls-native-tls"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }

[build-dependencies]
sqlx = { version = "0.8", features = ["postgres"] }

Setup

# Set DATABASE_URL for compile-time checking
export DATABASE_URL="postgresql://user:pass@localhost/db"

# Install SQLx CLI
cargo install sqlx-cli

# Create database and run migrations
sqlx database create
sqlx migrate add initial_schema
sqlx migrate run

Migrations

-- migrations/20231201000000_initial_schema.sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    name VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_users_email ON users(email);

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT,
    published BOOLEAN DEFAULT false,
    author_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Usage

use sqlx::postgres::PgPoolOptions;
use sqlx::{FromRow, PgPool};
use serde::{Deserialize, Serialize};

#[derive(Debug, FromRow, Serialize, Deserialize)]
struct User {
    id: i32,
    email: String,
    name: String,
}

#[derive(Debug, FromRow, Serialize, Deserialize)]
struct Post {
    id: i32,
    title: String,
    content: Option<String>,
    published: bool,
    author_id: i32,
}

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    // Create connection pool
    let pool = PgPoolOptions::new()
        .max_connections(20)
        .min_connections(5)
        .acquire_timeout(std::time::Duration::from_secs(30))
        .connect(&std::env::var("DATABASE_URL").unwrap())
        .await?;

    // Create
    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name"
    )
    .bind("test@example.com")
    .bind("Test User")
    .fetch_one(&pool)
    .await?;

    println!("Created user: {:?}", user);

    // Read - compile-time checked query!
    let users = sqlx::query_as::<_, User>(
        "SELECT id, email, name FROM users WHERE email LIKE $1"
    )
    .bind("%@example.com")
    .fetch_all(&pool)
    .await?;

    // Update
    sqlx::query("UPDATE users SET name = $1 WHERE id = $2")
        .bind("Updated Name")
        .bind(user.id)
        .execute(&pool)
        .await?;

    // Delete
    sqlx::query("DELETE FROM users WHERE id = $1")
        .bind(user.id)
        .execute(&pool)
        .await?;

    // Transaction
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO users (email, name) VALUES ($1, $2)")
        .bind("user1@example.com")
        .bind("User 1")
        .execute(&mut *tx)
        .await?;
    sqlx::query("INSERT INTO users (email, name) VALUES ($1, $2)")
        .bind("user2@example.com")
        .bind("User 2")
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;

    Ok(())
}

Compile-Time Checked Macros

// sqlx::query! macro - fully type-checked at compile time
let user = sqlx::query!(
    "SELECT id, email, name FROM users WHERE id = $1",
    user_id
)
.fetch_one(&pool)
.await?;

// Returns anonymous struct with fields:
// user.id: i32
// user.email: String
// user.name: String

// sqlx::query_as! - checked + custom struct
let user = sqlx::query_as!(
    User,
    "SELECT id, email, name FROM users WHERE email = $1",
    "test@example.com"
)
.fetch_one(&pool)
.await?;

Axum Integration

use axum::{
    extract::{Path, State},
    http::StatusCode,
    routing::{get, post},
    Json, Router,
};
use serde_json::json;
use sqlx::PgPool;

#[tokio::main]
async fn main() {
    let pool = PgPoolOptions::new()
        .max_connections(20)
        .connect(&std::env::var("DATABASE_URL").unwrap())
        .await
        .unwrap();

    let app = Router::new()
        .route("/users", post(create_user))
        .route("/users/:id", get(get_user))
        .with_state(pool);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn create_user(
    State(pool): State<PgPool>,
    Json(payload): Json<CreateUserPayload>,
) -> Result<Json<User>, StatusCode> {
    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name"
    )
    .bind(&payload.email)
    .bind(&payload.name)
    .fetch_one(&pool)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(user))
}

async fn get_user(
    State(pool): State<PgPool>,
    Path(id): Path<i32>,
) -> Result<Json<User>, StatusCode> {
    let user = sqlx::query_as::<_, User>(
        "SELECT id, email, name FROM users WHERE id = $1"
    )
    .bind(id)
    .fetch_one(&pool)
    .await
    .map_err(|_| StatusCode::NOT_FOUND)?;

    Ok(Json(user))
}

#[derive(Deserialize)]
struct CreateUserPayload {
    email: String,
    name: String,
}

Why SQLx?

• Compile-time query validation (catch errors before runtime!)
• Zero cost abstractions
• Async-first with Tokio
• PostgreSQL, MySQL, SQLite support
• Excellent performance

SeaORM (Full ORM)

Installation

[dependencies]
sea-orm = { version = "1", features = ["sqlx-postgres", "runtime-tokio-native-tls", "macros"] }
tokio = { version = "1", features = ["full"] }

Entity Definition

use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
#[sea_orm(table_name = "users")]
pub struct Model {
    #[sea_orm(primary_key)]
    pub id: i32,
    pub email: String,
    pub name: String,
}

#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {
    #[sea_orm(has_many = "super::post::Entity")]
    Post,
}

impl Related<super::post::Entity> for Entity {
    fn to() -> RelationDef {
        Relation::Post.def()
    }
}

impl ActiveModelBehavior for ActiveModel {}

Usage

use sea_orm::*;

#[tokio::main]
async fn main() -> Result<(), DbErr> {
    let db = Database::connect("postgresql://user:pass@localhost/db").await?;

    // Create
    let user = users::ActiveModel {
        email: Set("test@example.com".to_owned()),
        name: Set("Test User".to_owned()),
        ..Default::default()
    };
    let insert_result = user.insert(&db).await?;

    // Read
    let user = Users::find_by_id(1).one(&db).await?;

    let users = Users::find()
        .filter(users::Column::Email.contains("@example.com"))
        .order_by_asc(users::Column::Name)
        .all(&db)
        .await?;

    // Update
    let mut user: users::ActiveModel = user.unwrap().into();
    user.name = Set("Updated Name".to_owned());
    user.update(&db).await?;

    // Delete
    user.delete(&db).await?;

    // Relations
    let users_with_posts = Users::find()
        .find_with_related(Posts)
        .all(&db)
        .await?;

    Ok(())
}

Why SeaORM?

• Full ORM features (relations, migrations, CLI)
• Active Record + Data Mapper patterns
• Async-first
• Good documentation

Diesel (Mature, Stable)

Installation

[dependencies]
diesel = { version = "2.3", features = ["postgres", "r2d2"] }
diesel-async = { version = "0.5", features = ["postgres", "tokio"] }
dotenvy = "0.15"

Schema Definition

// src/schema.rs (generated by Diesel CLI)
diesel::table! {
    users (id) {
        id -> Int4,
        email -> Varchar,
        name -> Varchar,
    }
}

diesel::table! {
    posts (id) {
        id -> Int4,
        title -> Text,
        content -> Nullable<Text>,
        published -> Bool,
        author_id -> Int4,
    }
}

diesel::joinable!(posts -> users (author_id));
diesel::allow_tables_to_appear_in_same_query!(users, posts);

Usage

use diesel::prelude::*;
use diesel::r2d2::{self, ConnectionManager};

type Pool = r2d2::Pool<ConnectionManager<PgConnection>>;

#[derive(Queryable, Selectable)]
#[diesel(table_name = users)]
struct User {
    id: i32,
    email: String,
    name: String,
}

#[derive(Insertable)]
#[diesel(table_name = users)]
struct NewUser {
    email: String,
    name: String,
}

fn main() {
    let database_url = std::env::var("DATABASE_URL").unwrap();
    let manager = ConnectionManager::<PgConnection>::new(database_url);
    let pool = r2d2::Pool::builder().build(manager).unwrap();

    let mut conn = pool.get().unwrap();

    // Create
    let new_user = NewUser {
        email: "test@example.com".to_string(),
        name: "Test User".to_string(),
    };
    diesel::insert_into(users::table)
        .values(&new_user)
        .execute(&mut conn)
        .unwrap();

    // Read
    let results = users::table
        .filter(users::email.like("%@example.com"))
        .select(User::as_select())
        .load(&mut conn)
        .unwrap();

    // Update
    diesel::update(users::table.find(1))
        .set(users::name.eq("Updated Name"))
        .execute(&mut conn)
        .unwrap();

    // Delete
    diesel::delete(users::table.find(1))
        .execute(&mut conn)
        .unwrap();
}

Why Diesel?

• Mature and stable (oldest Rust ORM)
• Compile-time guarantees
• Excellent performance
• CLI for schema management

Comparison

Choose SQLx When:

• Want compile-time query validation
• Prefer writing SQL directly
• Need maximum performance
• Want minimal abstraction

Choose SeaORM When:

• Need full ORM features
• Want entity relations
• Prefer Active Record pattern
• Need migrations + CLI tools

Choose Diesel When:

• Need maximum stability
• Want compile-time type safety
• Prefer Data Mapper pattern
• Have complex schema

Best Practices

Connection Pooling

// SQLx
let pool = PgPoolOptions::new()
    .max_connections(20)
    .min_connections(5)
    .acquire_timeout(Duration::from_secs(30))
    .idle_timeout(Some(Duration::from_secs(600)))
    .max_lifetime(Some(Duration::from_secs(3600)))
    .connect(&database_url)
    .await?;

// Diesel
let manager = ConnectionManager::<PgConnection>::new(database_url);
let pool = r2d2::Pool::builder()
    .max_size(20)
    .connection_timeout(Duration::from_secs(30))
    .build(manager)?;

Error Handling

use thiserror::Error;

#[derive(Error, Debug)]
pub enum DatabaseError {
    #[error("Database error: {0}")]
    Sqlx(#[from] sqlx::Error),

    #[error("User not found")]
    NotFound,

    #[error("Duplicate email")]
    DuplicateEmail,
}

async fn get_user(pool: &PgPool, id: i32) -> Result<User, DatabaseError> {
    sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .map_err(|e| match e {
            sqlx::Error::RowNotFound => DatabaseError::NotFound,
            e => DatabaseError::Sqlx(e),
        })
}

Transactions

// SQLx
let mut tx = pool.begin().await?;
sqlx::query("INSERT INTO users (email, name) VALUES ($1, $2)")
    .bind("user1@example.com")
    .bind("User 1")
    .execute(&mut *tx)
    .await?;
sqlx::query("INSERT INTO users (email, name) VALUES ($1, $2)")
    .bind("user2@example.com")
    .bind("User 2")
    .execute(&mut *tx)
    .await?;
tx.commit().await?;

// SeaORM
let txn = db.begin().await?;
users::Entity::insert(user1).exec(&txn).await?;
users::Entity::insert(user2).exec(&txn).await?;
txn.commit().await?;

Resources

• SQLx: https://github.com/launchbadge/sqlx
• SeaORM: https://www.sea-ql.org/SeaORM/
• Diesel: https://diesel.rs/

TypeScript ORMs and Query Builders

Table of Contents

• Overview
• Prisma (Recommended for Best DX)
• Installation
• Schema Definition
• Generate Client
• Usage
• Next.js Integration
• Migrations
• Drizzle ORM (Recommended for Performance)
• Installation
• Schema Definition
• Usage
• Migrations
• Kysely (Type-Safe SQL Builder)
• Installation
• Schema Definition
• Usage
• TypeORM (Legacy)
• Comparison: When to Use Each
• Choose Prisma When:
• Choose Drizzle When:
• Choose Kysely When:
• Choose TypeORM When:
• Best Practices
• Connection Pooling for Serverless
• Type-Safe Queries
• Pagination
• Transactions
• Resources

Overview

Prisma (Recommended for Best DX)

Context7: /prisma/prisma (96.4 score, 4,281 doc snippets)

Installation

npm install prisma @prisma/client
npx prisma init

Schema Definition

// schema.prisma
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

generator client {
  provider = "prisma-client-js"
}

model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  name      String
  createdAt DateTime @default(now())
  posts     Post[]
  profile   Profile?

  @@index([email])
}

model Post {
  id        Int      @id @default(autoincrement())
  title     String
  content   String?
  published Boolean  @default(false)
  authorId  Int
  author    User     @relation(fields: [authorId], references: [id], onDelete: Cascade)

  @@index([authorId])
}

model Profile {
  id     Int    @id @default(autoincrement())
  bio    String
  userId Int    @unique
  user   User   @relation(fields: [userId], references: [id])
}

Generate Client

npx prisma migrate dev --name init  # Create migration + generate client
npx prisma generate  # Regenerate client after schema changes

Usage

import { PrismaClient } from '@prisma/client'

const prisma = new PrismaClient()

// Create
const user = await prisma.user.create({
  data: {
    email: 'test@example.com',
    name: 'Test User',
    posts: {
      create: [
        { title: 'First Post', content: 'Hello World' },
        { title: 'Second Post', content: 'More content' }
      ]
    }
  },
  include: { posts: true }  // Include relations
})

// Read
const user = await prisma.user.findUnique({
  where: { email: 'test@example.com' },
  include: { posts: true, profile: true }
})

const users = await prisma.user.findMany({
  where: { email: { contains: '@example.com' } },
  orderBy: { createdAt: 'desc' },
  take: 10,
  skip: 0
})

// Update
const updated = await prisma.user.update({
  where: { id: 1 },
  data: { name: 'Updated Name' }
})

// Delete
await prisma.user.delete({
  where: { id: 1 }
})

// Transactions
await prisma.$transaction([
  prisma.user.create({ data: { email: 'user1@example.com', name: 'User 1' } }),
  prisma.user.create({ data: { email: 'user2@example.com', name: 'User 2' } })
])

// Raw SQL
const users = await prisma.$queryRaw`SELECT * FROM users WHERE email = ${email}`

Next.js Integration

// lib/prisma.ts (connection pooling for serverless)
import { PrismaClient } from '@prisma/client'

const globalForPrisma = global as unknown as { prisma: PrismaClient }

export const prisma =
  globalForPrisma.prisma ||
  new PrismaClient({
    log: ['query', 'error', 'warn'],
  })

if (process.env.NODE_ENV !== 'production') globalForPrisma.prisma = prisma

// app/api/users/route.ts
import { NextResponse } from 'next/server'
import { prisma } from '@/lib/prisma'

export async function GET() {
  const users = await prisma.user.findMany()
  return NextResponse.json(users)
}

export async function POST(request: Request) {
  const body = await request.json()
  const user = await prisma.user.create({ data: body })
  return NextResponse.json(user)
}

Migrations

npx prisma migrate dev --name add_users  # Development
npx prisma migrate deploy  # Production
npx prisma migrate reset  # Reset database
npx prisma db push  # Prototype mode (skip migrations)

Why Prisma?

• Best-in-class TypeScript types (autocomplete for everything)
• Visual database browser (Prisma Studio: npx prisma studio)
• Migrations built-in
• Excellent documentation
• Large community

Drizzle ORM (Recommended for Performance)

Context7: /drizzle-team/drizzle-orm-docs (95.4 score, 4,037 snippets)

Installation

npm install drizzle-orm pg
npm install -D drizzle-kit

Schema Definition

// schema.ts
import { pgTable, serial, text, integer, boolean, timestamp } from 'drizzle-orm/pg-core'
import { relations } from 'drizzle-orm'

export const users = pgTable('users', {
  id: serial('id').primaryKey(),
  email: text('email').notNull().unique(),
  name: text('name').notNull(),
  createdAt: timestamp('created_at').defaultNow(),
})

export const posts = pgTable('posts', {
  id: serial('id').primaryKey(),
  title: text('title').notNull(),
  content: text('content'),
  published: boolean('published').default(false),
  authorId: integer('author_id').references(() => users.id, { onDelete: 'cascade' }),
})

// Define relations
export const usersRelations = relations(users, ({ many }) => ({
  posts: many(posts),
}))

export const postsRelations = relations(posts, ({ one }) => ({
  author: one(users, {
    fields: [posts.authorId],
    references: [users.id],
  }),
}))

Usage

import { drizzle } from 'drizzle-orm/node-postgres'
import { eq, like, desc, and, or } from 'drizzle-orm'
import { Pool } from 'pg'
import { users, posts } from './schema'

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
const db = drizzle(pool)

// Create
const newUser = await db.insert(users).values({
  email: 'test@example.com',
  name: 'Test User'
}).returning()

// Read
const allUsers = await db.select().from(users)

const user = await db.select().from(users).where(eq(users.id, 1))

const filtered = await db.select().from(users).where(
  and(
    like(users.email, '%@example.com'),
    eq(users.name, 'Test User')
  )
)

// Update
await db.update(users)
  .set({ name: 'Updated Name' })
  .where(eq(users.id, 1))

// Delete
await db.delete(users).where(eq(users.id, 1))

// Joins (type-safe!)
const usersWithPosts = await db
  .select()
  .from(users)
  .leftJoin(posts, eq(users.id, posts.authorId))

// Relations (automatic joins)
const usersWithPosts = await db.query.users.findMany({
  with: {
    posts: true,
  },
})

// Transactions
await db.transaction(async (tx) => {
  await tx.insert(users).values({ email: 'user1@example.com', name: 'User 1' })
  await tx.insert(users).values({ email: 'user2@example.com', name: 'User 2' })
})

Migrations

// drizzle.config.ts
import type { Config } from 'drizzle-kit'

export default {
  schema: './schema.ts',
  out: './drizzle',
  driver: 'pg',
  dbCredentials: {
    connectionString: process.env.DATABASE_URL!,
  },
} satisfies Config

npx drizzle-kit generate:pg  # Generate migrations
npx drizzle-kit push:pg  # Push schema directly (prototyping)
npx drizzle-kit studio  # Visual database browser

Why Drizzle?

• SQL-like syntax (easy migration from raw SQL)
• Zero runtime overhead (no proxies, minimal abstractions)
• ~14KB bundle size (vs Prisma's client generation)
• Fully type-safe with excellent inference
• Fast performance

Kysely (Type-Safe SQL Builder)

Installation

npm install kysely pg

Schema Definition

// types.ts
import { Generated, Insertable, Selectable, Updateable } from 'kysely'

export interface Database {
  users: UsersTable
  posts: PostsTable
}

export interface UsersTable {
  id: Generated<number>
  email: string
  name: string
  created_at: Generated<Date>
}

export interface PostsTable {
  id: Generated<number>
  title: string
  content: string | null
  published: Generated<boolean>
  author_id: number
}

export type User = Selectable<UsersTable>
export type NewUser = Insertable<UsersTable>
export type UserUpdate = Updateable<UsersTable>

Usage

import { Kysely, PostgresDialect } from 'kysely'
import { Pool } from 'pg'
import { Database } from './types'

const db = new Kysely<Database>({
  dialect: new PostgresDialect({
    pool: new Pool({ connectionString: process.env.DATABASE_URL })
  })
})

// Create
const user = await db
  .insertInto('users')
  .values({ email: 'test@example.com', name: 'Test User' })
  .returningAll()
  .executeTakeFirstOrThrow()

// Read
const users = await db.selectFrom('users').selectAll().execute()

const user = await db
  .selectFrom('users')
  .selectAll()
  .where('email', '=', 'test@example.com')
  .executeTakeFirst()

// Update
await db
  .updateTable('users')
  .set({ name: 'Updated Name' })
  .where('id', '=', 1)
  .execute()

// Delete
await db.deleteFrom('users').where('id', '=', 1).execute()

// Joins
const usersWithPosts = await db
  .selectFrom('users')
  .leftJoin('posts', 'posts.author_id', 'users.id')
  .select([
    'users.id',
    'users.email',
    'users.name',
    'posts.title',
    'posts.content'
  ])
  .execute()

// Transactions
await db.transaction().execute(async (trx) => {
  await trx.insertInto('users').values({ email: 'user1@example.com', name: 'User 1' }).execute()
  await trx.insertInto('users').values({ email: 'user2@example.com', name: 'User 2' }).execute()
})

Why Kysely?

• Pure SQL builder (no ORM magic)
• Excellent TypeScript inference
• Lightweight (~10KB)
• Database-agnostic (PostgreSQL, MySQL, SQLite)
• Type-safe query building

TypeORM (Legacy)

Use when: Maintaining existing TypeORM projects, prefer decorators.

import { Entity, PrimaryGeneratedColumn, Column, OneToMany, ManyToOne } from 'typeorm'

@Entity()
export class User {
  @PrimaryGeneratedColumn()
  id: number

  @Column({ unique: true })
  email: string

  @Column()
  name: string

  @OneToMany(() => Post, post => post.author)
  posts: Post[]
}

@Entity()
export class Post {
  @PrimaryGeneratedColumn()
  id: number

  @Column()
  title: string

  @Column({ nullable: true })
  content: string

  @ManyToOne(() => User, user => user.posts)
  author: User
}

Comparison: When to Use Each

Choose Prisma When:

• Want best developer experience
• Need visual database browser
• Prefer schema-first approach
• Building with Next.js/serverless
• Team new to databases

Choose Drizzle When:

• Performance critical
• Prefer SQL-like syntax
• Want minimal bundle size
• Comfortable with SQL
• Need full query control

Choose Kysely When:

• Want pure SQL builder
• Don't need ORM features
• Prefer explicit queries
• Type safety priority
• Lightweight solution

Choose TypeORM When:

• Maintaining legacy code
• Prefer decorators
• Like Active Record pattern
• Need repository pattern

Best Practices

Connection Pooling for Serverless

// Prisma (automatic pooling)
// Configure in DATABASE_URL:
// postgresql://user:pass@host/db?connection_limit=1&pool_timeout=30

// Drizzle + pg
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 1,  // Serverless: 1-2 connections per function
  idleTimeoutMillis: 30000,
})

// Use connection pooler (pgBouncer) for serverless
// DATABASE_URL=postgresql://user:pass@pooler-host:6543/db?pgbouncer=true

Type-Safe Queries

// Prisma (automatic)
const user = await prisma.user.findUnique({ where: { id: 1 } })
//    ^? User | null (inferred)

// Drizzle (automatic)
const users = await db.select().from(users)
//    ^? { id: number; email: string; name: string }[]

// Kysely (automatic)
const user = await db.selectFrom('users').selectAll().execute()
//    ^? User[]

Pagination

// Prisma
const users = await prisma.user.findMany({
  skip: (page - 1) * pageSize,
  take: pageSize,
  orderBy: { createdAt: 'desc' }
})

// Drizzle
const users = await db.select().from(users)
  .limit(pageSize)
  .offset((page - 1) * pageSize)
  .orderBy(desc(users.createdAt))

// Kysely
const users = await db.selectFrom('users')
  .selectAll()
  .limit(pageSize)
  .offset((page - 1) * pageSize)
  .orderBy('created_at', 'desc')
  .execute()

Transactions

// Prisma
await prisma.$transaction([
  prisma.user.create({ data: { email: 'user1@example.com', name: 'User 1' } }),
  prisma.post.create({ data: { title: 'Post 1', authorId: 1 } })
])

// Drizzle
await db.transaction(async (tx) => {
  await tx.insert(users).values({ email: 'user1@example.com', name: 'User 1' })
  await tx.insert(posts).values({ title: 'Post 1', authorId: 1 })
})

// Kysely
await db.transaction().execute(async (trx) => {
  await trx.insertInto('users').values({ email: 'user1@example.com', name: 'User 1' }).execute()
  await trx.insertInto('posts').values({ title: 'Post 1', author_id: 1 }).execute()
})

Resources

• Prisma Docs: https://www.prisma.io/docs
• Drizzle Docs: https://orm.drizzle.team/
• Kysely Docs: https://kysely.dev/
• TypeORM Docs: https://typeorm.io/

Serverless Databases Guide

Setup, branching workflows, and cost comparisons for Neon (PostgreSQL), PlanetScale (MySQL), and Turso (SQLite).

Table of Contents

• Comparison Overview
• Neon PostgreSQL
• Key Features
• Setup
• Connection String
• Database Branching Workflow
• Autoscaling Configuration
• Time-Travel Queries
• PlanetScale MySQL
• Key Features
• Setup
• Connection String
• Non-Blocking Schema Changes
• Prisma Integration
• Read Replicas
• Turso SQLite
• Key Features
• Setup
• Connection
• Embedded Replicas (Local-First)
• Edge Deployment
• Cost Comparison
• Neon (PostgreSQL)
• PlanetScale (MySQL)
• Turso (SQLite)
• Decision Framework
• Choose Neon if:
• Choose PlanetScale if:
• Choose Turso if:
• Integration Examples
• Next.js + Neon (Vercel)
• Cloudflare Workers + Turso
• Prisma + PlanetScale
• Best Practices
• Neon
• PlanetScale
• Turso
• Resources

Comparison Overview

---

Neon PostgreSQL

Best for: Preview environments, development branching, scale-to-zero PostgreSQL

Key Features

• Instant database branches - Create copy of production in <1 second
• Scale-to-zero compute - Automatically pause after inactivity
• Time-travel queries - Query database state from any point in time
• Autoscaling - Scale compute from 0.25 to 8 vCPU automatically
• Point-in-time restore - Restore to any second in the last 30 days

Setup

# Install Neon CLI
npm install -g neonctl

# Create project
neonctl projects create --name myapp

# Create branch
neonctl branches create --name dev --parent main

# Get connection string
neonctl connection-string main

Connection String

# Pooled connection (recommended for serverless)
postgresql://user:password@ep-cool-darkness-123456-pooler.us-east-2.aws.neon.tech/mydb?sslmode=require

# Direct connection (for migrations)
postgresql://user:password@ep-cool-darkness-123456.us-east-2.aws.neon.tech/mydb?sslmode=require

Database Branching Workflow

Feature development:

# 1. Create branch from main
neonctl branches create --name feature-auth

# 2. Get branch connection string
neonctl connection-string feature-auth

# 3. Run migrations on branch
DATABASE_URL=$(neonctl connection-string feature-auth) npm run prisma migrate dev

# 4. Test feature with real data (copy of production)
# 5. Delete branch when done
neonctl branches delete feature-auth

Preview deployments (Vercel integration):

# Each PR gets a database branch automatically
# vercel.json
{
  "build": {
    "env": {
      "DATABASE_URL": "@neon-preview-branch"
    }
  }
}

Autoscaling Configuration

// Neon automatically scales compute based on load
// Configure via dashboard or API:
// - Min: 0.25 vCPU (scale-to-zero)
// - Max: 8 vCPU
// - Autosuspend: 5 minutes of inactivity

Time-Travel Queries

-- Query database state from 2 hours ago
SELECT * FROM users AS OF SYSTEM TIME '2025-12-03 10:00:00';

-- Point-in-time restore via dashboard or CLI
neonctl branches restore main --timestamp '2025-12-03 10:00:00'

---

PlanetScale MySQL

Best for: Non-blocking schema changes, MySQL compatibility, read-heavy workloads

Key Features

• Non-blocking schema changes - Deploy schema changes with zero downtime
• Database branching - Create development branches with production schema
• Built-in read replicas - Automatic geographic distribution
• Online DDL - ALTER TABLE without locking
• No foreign key constraints - Application-enforced referential integrity

Setup

# Install PlanetScale CLI
brew install planetscale/tap/pscale

# Authenticate
pscale auth login

# Create database
pscale database create myapp --region us-east

# Create branch
pscale branch create myapp dev

Connection String

# Get connection string (includes SSL certificate)
pscale connect myapp main --port 3309

# Connection string format
mysql://username:password@aws.connect.psdb.cloud/myapp?ssl={"rejectUnauthorized":true}

Non-Blocking Schema Changes

Deploy request workflow:

# 1. Create development branch
pscale branch create myapp add-user-phone

# 2. Connect to branch
pscale shell myapp add-user-phone

# 3. Make schema changes
mysql> ALTER TABLE users ADD COLUMN phone VARCHAR(20);

# 4. Create deploy request
pscale deploy-request create myapp add-user-phone

# 5. Review diff
pscale deploy-request diff myapp 1

# 6. Deploy (zero downtime)
pscale deploy-request deploy myapp 1

# 7. Auto-merge to main after deploy

Prisma Integration

// schema.prisma
datasource db {
  provider = "mysql"
  url = env("DATABASE_URL")
  relationMode = "prisma"  // Required: No FK constraints
}

model User {
  id    Int    @id @default(autoincrement())
  email String @unique
  posts Post[]
}

model Post {
  id     Int    @id @default(autoincrement())
  userId Int
  user   User   @relation(fields: [userId], references: [id])

  @@index([userId])  // Manual index (replaces FK)
}

Read Replicas

// Automatically routed to nearest replica
const users = await prisma.user.findMany();  // Read from replica

// Force primary for consistency
const user = await prisma.user.create({ data: { email: 'test@example.com' } });  // Write to primary

---

Turso SQLite

Best for: Edge applications, local-first apps, global low-latency reads

Key Features

• Edge replicas - Deploy to 200+ locations globally
• Embedded replicas - Local SQLite + cloud sync
• Sub-10ms reads - Query from nearest edge location
• LibSQL - SQLite fork with extensions
• Multi-region writes - Eventually consistent replication

Setup

# Install Turso CLI
curl -sSfL https://get.tur.so/install.sh | bash

# Create database
turso db create myapp

# Create replica in multiple regions
turso db replicate myapp --region ams  # Amsterdam
turso db replicate myapp --region sin  # Singapore

# Get connection URL
turso db show myapp --url

Connection

import { createClient } from '@libsql/client';

const db = createClient({
  url: process.env.TURSO_URL,
  authToken: process.env.TURSO_AUTH_TOKEN,
});

const users = await db.execute('SELECT * FROM users');

Embedded Replicas (Local-First)

import { createClient } from '@libsql/client';

const db = createClient({
  url: 'file:local.db',  // Local SQLite file
  syncUrl: process.env.TURSO_URL,  // Turso cloud
  authToken: process.env.TURSO_AUTH_TOKEN,
});

// Read from local (instant)
const users = await db.execute('SELECT * FROM users');

// Sync to cloud (background)
await db.sync();

Edge Deployment

// Cloudflare Worker + Turso
export default {
  async fetch(request, env) {
    const db = createClient({
      url: env.TURSO_URL,
      authToken: env.TURSO_AUTH_TOKEN,
    });

    const users = await db.execute('SELECT * FROM users');
    return Response.json(users.rows);
  },
};

---

Cost Comparison

Neon (PostgreSQL)

Free Tier:

• 0.5 GB storage
• 3 compute branches
• Shared compute (0.25 vCPU)

Launch ($19/month):

• 10 GB storage
• 10 branches
• 2 vCPU autoscaling (0.25-2)
• Point-in-time restore (7 days)

Scale ($69/month):

• 50 GB storage
• Unlimited branches
• 8 vCPU autoscaling (0.25-8)
• Point-in-time restore (30 days)

Break-even: 24/7 usage vs managed PostgreSQL (~$50/month)

---

PlanetScale (MySQL)

Hobby (Free):

• 5 GB storage
• 1 billion row reads/month
• 10 million row writes/month
• 1 production branch

Scaler ($29/month):

• 10 GB storage
• 100 billion row reads/month
• 50 million row writes/month
• 2 production branches
• Deploy requests

Business ($39/month per branch):

• Unlimited reads/writes
• Multiple production branches
• SSO/SAML

Break-even: High read volume vs managed MySQL

---

Turso (SQLite)

Starter (Free):

• 8 GB storage
• 500 database locations (replicas)
• 1 billion row reads/month

Scaler ($29/month):

• 50 GB storage
• 1,000 locations
• 100 billion row reads/month

Enterprise (Custom):

• Unlimited
• SLA guarantees
• Dedicated support

Break-even: Edge use cases with global distribution

---

Decision Framework

Choose Neon if:

• Need PostgreSQL features (JSONB, arrays, extensions)
• Want scale-to-zero compute
• Database branching for preview environments is critical
• Development team needs isolated database copies

Choose PlanetScale if:

• MySQL compatibility required
• Non-blocking schema changes are critical
• High read volume with geographic distribution
• Want automated read replicas

Choose Turso if:

• Building edge applications (Cloudflare Workers, Deno Deploy)
• Need sub-10ms global reads
• Local-first architecture (offline-capable apps)
• SQLite compatibility required

---

Integration Examples

Next.js + Neon (Vercel)

# 1. Create Neon database
neonctl projects create

# 2. Add to Vercel
vercel env add DATABASE_URL

# 3. Automatic branching (preview deployments)
# Each PR gets a Neon branch automatically

Cloudflare Workers + Turso

export default {
  async fetch(request, env) {
    const db = createClient({
      url: env.TURSO_URL,
      authToken: env.TURSO_AUTH_TOKEN,
    });

    const result = await db.execute({
      sql: 'SELECT * FROM users WHERE id = ?',
      args: [1],
    });

    return Response.json(result.rows[0]);
  },
};

Prisma + PlanetScale

// prisma/schema.prisma
datasource db {
  provider = "mysql"
  url = env("DATABASE_URL")
  relationMode = "prisma"
}

// Deploy schema changes
pscale branch create myapp migration
pscale connect myapp migration --port 3309
DATABASE_URL="mysql://127.0.0.1:3309/myapp" npx prisma db push
pscale deploy-request create myapp migration

---

Best Practices

Neon

1. Use pooled connections for serverless functions 2. Branch per feature for isolated development 3. Enable autoscaling for variable workloads 4. Use time-travel for debugging production issues

PlanetScale

1. Use deploy requests for all schema changes 2. Index all foreign key columns (no FK constraints) 3. Test schema changes on development branches 4. Monitor row read/write quotas on dashboard

Turso

1. Deploy replicas near users for low latency 2. Use embedded replicas for offline-first apps 3. Sync periodically in background (embedded mode) 4. Monitor replication lag in multi-region setups

---

Resources

• Neon Docs: https://neon.tech/docs
• PlanetScale Docs: https://planetscale.com/docs
• Turso Docs: https://docs.turso.tech