What is database-optimization?

The Database Optimization skill is designed to enhance database performance and reliability by analyzing SQL queries, identifying N+1 efficiency bottlenecks, and providing expert index recommendations. It specializes in SQLAlchemy async environments, offering safety reviews for Alembic migrations and actionable query execution plans to ensure scalable and high-performing data layers.

When should I use database-optimization?

database-optimization is useful in the following scenarios: • Performance Bottleneck Analysis: Identify and optimize slow-running SQL queries by analyzing execution plans and suggesting high-efficiency query patterns. • Index Strategy Optimization: Audit database schemas to recommend missing indexes for JOIN/WHERE clauses and identify redundant indexes to reduce write overhead. • Migration Safety Audits: Review Alembic migration scripts to detect high-risk operations like table locks or breaking changes before they reach production. • N+1 Problem Resolution: Automatically detect inefficient loop-based database queries in SQLAlchemy and provide eager-loading solutions using selectinload or joinedload. • Async Database Best Practices: Implement proper asynchronous session handling and bulk operation patterns to maximize throughput in modern Python web applications.

name	database-optimization
description	Optimize SQL queries, analyze indexes, review Alembic migrations, and identify N+1 problems. Provides query execution plans, index recommendations, and migration best practices for SQLAlchemy async.

Database Optimization Skill

When to Use

Use this skill when:

Optimizing slow SQL queries
Analyzing missing or redundant indexes
Reviewing Alembic migrations for safety
Identifying N+1 query problems
Designing database schemas
Planning data migrations
Reviewing query execution plans

Optimization Checklist

1. Index Analysis

-- Check missing indexes
SELECT schemaname, tablename, indexname, indexdef
FROM pg_indexes
WHERE tablename = 'your_table';

-- Identify unused indexes
SELECT * FROM pg_stat_user_indexes
WHERE idx_scan = 0;

Index Recommendations:

✅ Always index:
- Primary keys (automatic)
- Foreign keys
- Columns in WHERE clauses
- Columns in ORDER BY
- Columns in JOIN conditions

❌ Avoid indexing:
- Low cardinality columns (boolean, status)
- Frequently updated columns
- Small tables (<1000 rows)

2. N+1 Query Detection

# ❌ N+1 Problem
users = await session.execute(select(User))
for user in users:
    # This triggers N additional queries!
    orders = await session.execute(
        select(Order).where(Order.user_id == user.id)
    )

# ✅ Solution: Eager loading
from sqlalchemy.orm import selectinload

users = await session.execute(
    select(User).options(selectinload(User.orders))
)

3. Query Optimization Patterns

# ✅ Use pagination
from sqlalchemy import select
from app.models import Product

async def get_products(skip: int = 0, limit: int = 20):
    query = select(Product).offset(skip).limit(limit)
    result = await session.execute(query)
    return result.scalars().all()

# ✅ Use specific columns (not SELECT *)
query = select(Product.id, Product.name, Product.price)

# ✅ Use exists() for existence checks
from sqlalchemy import exists
query = select(exists().where(User.email == email))

4. Migration Safety

# ✅ Safe migration patterns
def upgrade():
    # Add column with default (no table lock)
    op.add_column('users', sa.Column('status', sa.String(20), 
                                      server_default='active'))

# ❌ Dangerous patterns
def upgrade():
    # Avoid: Rename column (breaks app)
    op.alter_column('users', 'name', new_column_name='full_name')
    
    # Avoid: Change column type (data loss risk)
    op.alter_column('users', 'age', type_=sa.String())

5. Async Best Practices

# ✅ Proper async session handling
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user(session: AsyncSession, user_id: int):
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

# ✅ Bulk operations
async def bulk_insert(session: AsyncSession, items: list):
    session.add_all(items)
    await session.commit()

Output Format

## Database Optimization Report

### Query Analysis
- Queries analyzed: X
- Slow queries (>100ms): X
- N+1 problems: X

### Index Recommendations
| Table | Column | Type | Reason |
|-------|--------|------|--------|
| users | email | UNIQUE | WHERE clause filter |
| orders | user_id | INDEX | Foreign key JOIN |

### Optimization Suggestions
1. [Query] - [Current time] - [Optimized time] - [How]

### Migration Review
- ✅ Safe to run
- ⚠️ Requires maintenance window
- ❌ Breaking change detected

Example Usage

@database Optimize the slow query in order_service.py:120
@database Review the new migration for safety issues
@database Find N+1 problems in the user module
@database Suggest indexes for the products table

database-optimization

When & Why to Use This Skill

Use Cases