# Schema Mapper

> Database schema understanding and relationship mapping. Use when exploring unfamiliar databases, documenting table relationships, identifying join paths, or generating ERD documentation for existing schemas.

- Skill: `nimrodfisher/schema-mapper` (Agent Skill)
- Install (CLI): `npx skillmds@latest add nimrodfisher/schema-mapper`
- Raw SKILL.md: https://api.skillmd.com/api/skills/nimrodfisher/schema-mapper/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Docs & Writing
- Author: nimrodfisher (https://skillmd.com/u/nimrodfisher)
- Updated: 2026-09-10
- Page: https://skillmd.com/skills/nimrodfisher/schema-mapper

---


# Schema Mapper

## Quick Start

Automatically discover, document, and visualize database schemas including tables, columns, relationships, and join paths. Essential for understanding unfamiliar databases or creating documentation.

## Context Requirements

Before mapping the schema, I need:

1. **Database Access**: Connection details or schema export
2. **Scope**: Which tables/schemas to map (or all)
3. **Documentation Goal**: What you need (ERD, join paths, data dictionary, etc.)
4. **Known Relationships** (optional): Explicit foreign keys or implicit relationships

## Context Gathering

### For Database Access:
"I can map your schema from:

**Option 1 - Direct Database Connection:**
```python
connection_details = {
    'host': 'your-db.example.com',
    'database': 'production',
    'user': 'readonly_user',
    'password': '***',
    'type': 'postgresql'  # or mysql, snowflake, bigquery, redshift
}
```

**Option 2 - Schema Export:**
```sql
-- PostgreSQL:
SELECT table_schema, table_name, column_name, data_type, 
       is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;

-- Also export constraints:
SELECT * FROM information_schema.table_constraints;
SELECT * FROM information_schema.key_column_usage;
```

**Option 3 - dbt Project:**
Share your `dbt_project.yml` and `models/` directory - I'll extract schema from dbt docs

**Option 4 - Existing Documentation:**
Share any ERDs, data dictionaries, or schema docs you have"

### For Scope:
"Should I map:
- **All tables** in the database?
- **Specific schema** (e.g., 'public', 'analytics', 'staging')?
- **Specific tables** (list the important ones)?
- **Tables matching pattern** (e.g., all 'fct_*' and 'dim_*' tables)?"

### For Documentation Goal:
"What do you need from this schema mapping?

**Common Goals:**
1. **Visual ERD** - See table relationships graphically
2. **Join Path Finder** - How to join Table A to Table B
3. **Data Dictionary** - Complete catalog of tables and columns
4. **Lineage Map** - Understand data flow through tables
5. **Quick Reference** - Cheat sheet for common joins

Which would be most useful?"

### For Relationships:
"I'll auto-detect relationships from:
- Foreign key constraints
- Column name patterns (id, user_id, etc.)
- Naming conventions

If you have implicit relationships (not enforced by FK constraints), let me know:
- Example: 'orders.customer_email relates to customers.email but no FK exists'"

## Workflow

### Step 1: Connect and Discover Schema

```python
import pandas as pd
import sqlalchemy
from sqlalchemy import create_engine, inspect

# Connect to database
engine = create_engine(connection_string)
inspector = inspect(engine)

# Get all schemas
schemas = inspector.get_schema_names()
print(f"📚 Schemas found: {schemas}")

# Get tables in target schema
tables = inspector.get_table_names(schema='public')
print(f"📊 Tables in 'public': {len(tables)}")
for table in sorted(tables):
    print(f"  - {table}")
```

**Checkpoint**: "Found {N} tables in schema. Does this look right? Any tables missing or unexpected?"

### Step 2: Extract Table Metadata

```python
def extract_table_metadata(engine, schema, table_name):
    """Extract comprehensive metadata for a single table"""
    inspector = inspect(engine)
    
    # Get columns
    columns = inspector.get_columns(table_name, schema=schema)
    
    # Get primary keys
    pk = inspector.get_pk_constraint(table_name, schema=schema)
    pk_columns = pk.get('constrained_columns', [])
    
    # Get foreign keys
    fks = inspector.get_foreign_keys(table_name, schema=schema)
    
    # Get indexes
    indexes = inspector.get_indexes(table_name, schema=schema)
    
    # Get unique constraints
    unique = inspector.get_unique_constraints(table_name, schema=schema)
    
    return {
        'table_name': table_name,
        'schema': schema,
        'columns': columns,
        'primary_keys': pk_columns,
        'foreign_keys': fks,
        'indexes': indexes,
        'unique_constraints': unique
    }

# Extract metadata for all tables
schema_metadata = {}
for table in tables:
    schema_metadata[table] = extract_table_metadata(engine, 'public', table)
    print(f"✓ Extracted metadata for {table}")
```

### Step 3: Infer Relationships

```python
def infer_relationships(schema_metadata):
    """
    Infer relationships from:
    1. Explicit foreign keys
    2. Column naming patterns (user_id → users.id)
    3. Common patterns (created_by → users.id)
    """
    relationships = []
    
    for table_name, metadata in schema_metadata.items():
        # Explicit FKs
        for fk in metadata['foreign_keys']:
            relationships.append({
                'type': 'explicit_fk',
                'from_table': table_name,
                'from_column': fk['constrained_columns'][0],
                'to_table': fk['referred_table'],
                'to_column': fk['referred_columns'][0],
                'confidence': 'high'
            })
        
        # Inferred from naming patterns
        for col in metadata['columns']:
            col_name = col['name']
            
            # Pattern: table_id → table.id
            if col_name.endswith('_id') and col_name != 'id':
                potential_table = col_name[:-3] + 's'  # users_id → users
                if potential_table in schema_metadata:
                    relationships.append({
                        'type': 'inferred_naming',
                        'from_table': table_name,
                        'from_column': col_name,
                        'to_table': potential_table,
                        'to_column': 'id',
                        'confidence': 'medium'
                    })
    
    return relationships

relationships = infer_relationships(schema_metadata)
print(f"🔗 Found {len(relationships)} relationships")
```

### Step 4: Generate Data Dictionary

```python
def generate_data_dictionary(schema_metadata):
    """Create comprehensive data dictionary"""
    
    dictionary = []
    
    for table_name, metadata in schema_metadata.items():
        for col in metadata['columns']:
            dictionary.append({
                'table': table_name,
                'column': col['name'],
                'type': str(col['type']),
                'nullable': col['nullable'],
                'default': col.get('default'),
                'is_pk': col['name'] in metadata['primary_keys'],
                'is_fk': any(col['name'] in fk['constrained_columns'] 
                           for fk in metadata['foreign_keys'])
            })
    
    df_dict = pd.DataFrame(dictionary)
    return df_dict

data_dict = generate_data_dictionary(schema_metadata)
print(f"\n📖 Data Dictionary:")
print(data_dict.head(20))

# Save to CSV
data_dict.to_csv('data_dictionary.csv', index=False)
```

### Step 5: Find Join Paths

```python
def find_join_path(from_table, to_table, relationships):
    """
    Find how to join from_table to to_table
    Using BFS to find shortest path
    """
    from collections import deque
    
    # Build adjacency graph
    graph = {}
    for rel in relationships:
        if rel['from_table'] not in graph:
            graph[rel['from_table']] = []
        graph[rel['from_table']].append({
            'to_table': rel['to_table'],
            'from_col': rel['from_column'],
            'to_col': rel['to_column'],
            'type': rel['type']
        })
    
    # BFS to find path
    queue = deque([(from_table, [])])
    visited = {from_table}
    
    while queue:
        current_table, path = queue.popleft()
        
        if current_table == to_table:
            return path
        
        if current_table in graph:
            for edge in graph[current_table]:
                if edge['to_table'] not in visited:
                    visited.add(edge['to_table'])
                    new_path = path + [edge]
                    queue.append((edge['to_table'], new_path))
    
    return None  # No path found

# Example: How to join orders to customers?
path = find_join_path('orders', 'customers', relationships)

if path:
    print("\n🛤️  Join Path: orders → customers")
    for i, step in enumerate(path, 1):
        print(f"  Step {i}: JOIN {step['to_table']} " +
              f"ON {path[i-1]['to_table'] if i > 1 else 'orders'}.{step['from_col']} = " +
              f"{step['to_table']}.{step['to_col']}")
else:
    print("❌ No join path found")
```

### Step 6: Generate ERD

```python
def generate_erd_mermaid(schema_metadata, relationships):
    """Generate Mermaid ERD diagram"""
    
    mermaid = ["erDiagram"]
    
    # Add tables with columns
    for table_name, metadata in schema_metadata.items():
        mermaid.append(f"    {table_name.upper()} {{")
        
        for col in metadata['columns'][:10]:  # Limit to key columns
            col_type = str(col['type'])[:20]
            pk_marker = " PK" if col['name'] in metadata['primary_keys'] else ""
            fk_marker = " FK" if any(col['name'] in fk['constrained_columns'] 
                                    for fk in metadata['foreign_keys']) else ""
            
            mermaid.append(f"        {col_type} {col['name']}{pk_marker}{fk_marker}")
        
        mermaid.append("    }")
    
    # Add relationships
    for rel in relationships:
        if rel['type'] == 'explicit_fk':
            mermaid.append(
                f"    {rel['from_table'].upper()} ||--o{{ " +
                f"{rel['to_table'].upper()} : {rel['from_column']}"
            )
    
    return "\n".join(mermaid)

erd = generate_erd_mermaid(schema_metadata, relationships)
print("\n📊 ERD (Mermaid format):")
print(erd)

# Save to file
with open('schema_erd.mmd', 'w') as f:
    f.write(erd)
```

### Step 7: Generate Quick Reference Guide

```python
def generate_quick_reference(schema_metadata, relationships):
    """Create quick reference for common joins"""
    
    ref = []
    ref.append("# Schema Quick Reference\n")
    
    # Table summary
    ref.append("## Tables Overview\n")
    for table_name, metadata in sorted(schema_metadata.items()):
        row_count = f"(~{metadata.get('row_count', '?')} rows)" if 'row_count' in metadata else ""
        ref.append(f"- **{table_name}** {row_count}")
        ref.append(f"  - Primary Key: {', '.join(metadata['primary_keys']) or 'None'}")
        ref.append(f"  - Columns: {len(metadata['columns'])}\n")
    
    # Common joins
    ref.append("\n## Common Join Patterns\n")
    
    # Group relationships by from_table
    joins_by_table = {}
    for rel in relationships:
        if rel['from_table'] not in joins_by_table:
            joins_by_table[rel['from_table']] = []
        joins_by_table[rel['from_table']].append(rel)
    
    for table, rels in sorted(joins_by_table.items()):
        ref.append(f"\n### From {table}:\n")
        for rel in rels:
            ref.append(
                f"```sql\n"
                f"JOIN {rel['to_table']} ON {table}.{rel['from_column']} = " +
                f"{rel['to_table']}.{rel['to_column']}\n"
                f"```\n"
            )
    
    return "\n".join(ref)

quick_ref = generate_quick_reference(schema_metadata, relationships)
print(quick_ref)

# Save to markdown
with open('schema_quick_reference.md', 'w') as f:
    f.write(quick_ref)
```

## Context Validation

Before proceeding, verify:
- [ ] Database connection works or schema export is complete
- [ ] Target tables/schemas are clearly defined
- [ ] Have permissions to query INFORMATION_SCHEMA
- [ ] Understand which relationships are explicit vs inferred
- [ ] Know what documentation format is needed

## Output Template

```
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
SCHEMA MAPPING REPORT
Database: production_db
Schema: public
Generated: 2025-01-11
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

📊 SCHEMA OVERVIEW
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Total Tables: 42
Total Columns: 387
Total Relationships: 56
  - Explicit (FK): 38
  - Inferred: 18

🗂️  TABLE CATEGORIES
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Fact Tables (fct_*): 8
Dimension Tables (dim_*): 12
Staging Tables (stg_*): 15
Raw Tables (raw_*): 7

📋 KEY TABLES
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

1. users (dim_users)
   - PK: id
   - Columns: 15
   - Referenced by: orders, sessions, events

2. orders (fct_orders)
   - PK: id
   - FK: user_id → users.id, product_id → products.id
   - Columns: 23

3. products (dim_products)
   - PK: id
   - Columns: 12
   - Referenced by: orders, cart_items

🔗 RELATIONSHIP MAP
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

orders → users (user_id)
orders → products (product_id)  
sessions → users (user_id)
events → sessions (session_id)
cart_items → products (product_id)

🛤️  COMMON JOIN PATHS
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

users → orders:
  JOIN orders ON users.id = orders.user_id

orders → products:
  JOIN products ON orders.product_id = products.id

users → events (via sessions):
  JOIN sessions ON users.id = sessions.user_id
  JOIN events ON sessions.id = events.session_id

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
FILES GENERATED
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

✓ schema_erd.mmd (Mermaid ERD)
✓ data_dictionary.csv (All columns)
✓ schema_quick_reference.md (Join guide)
✓ relationship_graph.json (Machine-readable)
```

## Common Scenarios

### Scenario 1: "New to the database, need overview"
→ Full schema map with ERD and data dictionary
→ Highlight fact and dimension tables
→ Show most common join patterns

### Scenario 2: "How do I join Table A to Table B?"
→ Use find_join_path to show exact SQL
→ If no direct path, show multi-hop joins
→ Validate path makes business sense

### Scenario 3: "Document schema for new team members"
→ Generate comprehensive quick reference
→ Include business context for key tables
→ Create visual ERD for overview

### Scenario 4: "Find all tables related to users"
→ Traverse relationship graph from users table
→ Show both direct and indirect relationships
→ Categorize by relationship type

### Scenario 5: "Validate schema against expectations"
→ Compare actual schema to documented schema
→ Flag missing tables, columns, relationships
→ Identify schema drift

## Handling Missing Context

**User provides database but no context:**
"I'll map the entire schema and present an overview. You can then tell me which areas to explore in depth."

**User has partial access (read-only):**
"I'll work with INFORMATION_SCHEMA queries which don't require write permissions. This will show structure but not data samples."

**User has dbt project instead of database:**
"Great! I'll extract schema from your dbt models and documentation. This often has better business context than raw database schema."

**User wants specific tables only:**
"I'll focus on those tables and map their direct relationships. Let me know if you want me to expand to related tables."

**Foreign keys not enforced:**
"I'll infer relationships from column naming patterns. Review the 'inferred' relationships and confirm which are correct."

## Advanced Options

After basic schema mapping, offer:

**Lineage Tracking**:
"Want me to trace data lineage? I can show how data flows from raw tables through transformations to final fact tables."

**Schema Comparison**:
"I can compare schemas across environments (dev vs prod) or over time to identify changes."

**Query Pattern Analysis**:
"If you provide query logs, I can identify most-used join patterns and optimize documentation around them."

**dbt Integration**:
"I can generate dbt schema.yml files from this mapping to document your models."

**Auto-Documentation**:
"I can set up automated schema documentation that updates when schema changes."

**Performance Insights**:
"Based on table sizes and join patterns, I can identify potential performance bottlenecks."