Skip to content

Core Database Tools

Jump to bottom
Temp edited this page Oct 3, 2025 · 1 revision

Core Database Tools

Essential PostgreSQL operations for schema management, SQL execution, and database health monitoring.

📊 Overview

9 core tools providing fundamental database operations:

Tool Purpose Security
list_schemas List all database schemas ✅ Safe
list_objects List tables, views, sequences, extensions ✅ Safe
get_object_details Detailed object information and schema ✅ Safe
execute_sql Secure SQL execution with parameter binding ✅ Protected
explain_query Query execution plans with cost analysis ✅ Safe
get_top_queries Real-time query performance analysis ✅ Safe
analyze_workload_indexes Workload-based index recommendations ✅ Safe
analyze_query_indexes Query-specific index optimization ✅ Safe
analyze_db_health Comprehensive database health checks ✅ Safe

🔧 Tool Details

list_schemas

List all database schemas with ownership and type information.

Parameters: None

Returns:

  • schema_name: Name of the schema
  • schema_owner: Owner of the schema
  • schema_type: Type (User Schema, System Schema, etc.)

Example:

result = list_schemas()
# Returns: [
#   {"schema_name": "public", "schema_owner": "postgres", "schema_type": "User Schema"},
#   {"schema_name": "pg_catalog", "schema_owner": "postgres", "schema_type": "System Schema"}
# ]

Use Cases:

  • Database exploration
  • Permission auditing
  • Schema management

list_objects

List database objects (tables, views, sequences, extensions) in a specific schema.

Parameters:

  • schema_name (string, required): Schema to list objects from
  • object_type (string, optional): Filter by type (table, view, sequence, extension)

Returns:

  • schema: Schema name
  • name: Object name
  • type: Object type

Example:

# List all tables in public schema
result = list_objects(
    schema_name="public",
    object_type="table"
)
# Returns: [
#   {"schema": "public", "name": "users", "type": "BASE TABLE"},
#   {"schema": "public", "name": "orders", "type": "BASE TABLE"}
# ]

Use Cases:

  • Schema exploration
  • Object inventory
  • Migration planning

get_object_details

Get detailed information about a specific database object including columns, constraints, and indexes.

Parameters:

  • schema_name (string, required): Schema containing the object
  • object_name (string, required): Name of the object
  • object_type (string, required): Type of object (table, view, sequence)

Returns:

  • basic: Object metadata
  • columns: Column definitions with data types
  • constraints: Primary keys, foreign keys, checks
  • indexes: Index definitions and types

Example:

result = get_object_details(
    schema_name="public",
    object_name="users",
    object_type="table"
)
# Returns: {
#   "basic": {"schema": "public", "name": "users", "type": "table"},
#   "columns": [
#     {"column": "id", "data_type": "integer", "is_nullable": "NO"},
#     {"column": "email", "data_type": "varchar", "is_nullable": "NO"}
#   ],
#   "constraints": [
#     {"name": "users_pkey", "type": "PRIMARY KEY", "columns": ["id"]}
#   ],
#   "indexes": [
#     {"name": "idx_users_email", "definition": "CREATE INDEX..."}
#   ]
# }

Use Cases:

  • Schema documentation
  • Migration planning
  • Performance analysis

execute_sql

Execute SQL queries with secure parameter binding to prevent SQL injection.

Parameters:

  • sql (string, required): SQL query with %s placeholders
  • params (array, optional): Parameters to bind to the query

Returns: Query results as array of objects

Example:

# ✅ SECURE: Parameter binding
result = execute_sql(
    sql="SELECT * FROM users WHERE id = %s AND active = %s",
    params=[123, True]
)

# ✅ SECURE: INSERT with parameters
result = execute_sql(
    sql="INSERT INTO products (name, price) VALUES (%s, %s)",
    params=["Widget", 29.99]
)

Security:

  • ✅ SQL injection prevention via parameter binding
  • ✅ Automatic query validation in restricted mode
  • ⚠️ Full write access in unrestricted mode

Use Cases:

  • Data querying
  • Data modification (unrestricted mode)
  • Complex analytics

explain_query

Analyze query execution plans with optional hypothetical index testing.

Parameters:

  • sql (string, required): SQL query to analyze
  • params (array, optional): Query parameters
  • analyze (boolean, optional): Execute query and get actual statistics
  • hypothetical_indexes (array, optional): Test indexes without creating them

Returns:

  • Formatted execution plan
  • Cost estimates
  • Performance recommendations

Example:

# Basic EXPLAIN
result = explain_query(
    sql="SELECT * FROM users WHERE email = %s",
    params=["user@example.com"],
    analyze=False
)

# Test hypothetical index (requires hypopg extension)
result = explain_query(
    sql="SELECT * FROM orders WHERE customer_id = %s",
    params=[123],
    hypothetical_indexes=[
        {"table": "orders", "columns": ["customer_id"], "using": "btree"}
    ]
)

Use Cases:

  • Query optimization
  • Index planning
  • Performance tuning

get_top_queries

Get real-time query performance statistics using pg_stat_statements extension.

Parameters:

  • sort_by (string, optional): Sort criteria (total_time, mean_time, calls)
  • limit (integer, optional): Number of queries to return (default: 10)

Returns:

  • query: SQL query text
  • calls: Number of executions
  • total_exec_time: Total execution time (ms)
  • mean_exec_time: Average execution time (ms)
  • rows: Total rows returned

Example:

# Get slowest queries by total time
result = get_top_queries(
    sort_by="total_time",
    limit=10
)

# Get most frequently called queries
result = get_top_queries(
    sort_by="calls",
    limit=5
)

Requirements: pg_stat_statements extension must be installed

Use Cases:

  • Performance monitoring
  • Query optimization
  • Workload analysis

analyze_workload_indexes

Analyze current workload and recommend indexes using DTA (Database Tuning Advisor) algorithm.

Parameters:

  • method (string, optional): Analysis method (dta, basic)
  • max_index_size_mb (integer, optional): Maximum index size to recommend

Returns:

  • Recommended indexes with estimated benefits
  • Cost-benefit analysis
  • Implementation SQL

Example:

result = analyze_workload_indexes(
    method="dta",
    max_index_size_mb=1000
)
# Returns: {
#   "recommendations": [
#     {
#       "table": "orders",
#       "columns": ["customer_id", "order_date"],
#       "estimated_benefit": "45% improvement",
#       "estimated_size_mb": 125,
#       "create_sql": "CREATE INDEX..."
#     }
#   ]
# }

Requirements: pg_stat_statements for workload analysis, hypopg for simulation

Use Cases:

  • Performance optimization
  • Index strategy planning
  • Database tuning

analyze_query_indexes

Analyze specific queries and recommend optimal indexes.

Parameters:

  • queries (array, required): List of SQL queries to analyze
  • method (string, optional): Analysis method (dta, basic)

Returns:

  • Query-specific index recommendations
  • Performance impact estimates
  • Implementation SQL

Example:

result = analyze_query_indexes(
    queries=[
        "SELECT * FROM users WHERE email = %s",
        "SELECT * FROM orders WHERE customer_id = %s AND status = %s"
    ],
    method="dta"
)

Use Cases:

  • Query optimization
  • Targeted performance improvements
  • Index planning

analyze_db_health

Comprehensive database health analysis covering indexes, connections, vacuum, buffers, and more.

Parameters:

  • health_type (string, optional): Type of check (all, index, connection, vacuum, buffer, replication, constraint)

Returns: Detailed health report with issues and recommendations

Example:

# Comprehensive health check
result = analyze_db_health(health_type="all")

# Specific health areas
result = analyze_db_health(health_type="index")     # Index bloat and usage
result = analyze_db_health(health_type="buffer")    # Cache hit rates
result = analyze_db_health(health_type="vacuum")    # Transaction wraparound
result = analyze_db_health(health_type="connection") # Connection pool status

Health Checks:

  • ✅ Invalid Indexes - Detect and report corrupt indexes
  • ✅ Duplicate Indexes - Find redundant indexes
  • ✅ Index Bloat - Identify bloated indexes
  • ✅ Unused Indexes - Find rarely-used indexes
  • ✅ Connection Health - Monitor connection pool utilization
  • ✅ Vacuum Health - Prevent transaction ID wraparound
  • ✅ Sequence Health - Monitor sequence usage
  • ✅ Replication Health - Check replication lag
  • ✅ Buffer Health - Analyze cache hit rates (99%+ accuracy)
  • ✅ Constraint Health - Detect invalid constraints

Use Cases:

  • Proactive database maintenance
  • Performance monitoring
  • Problem detection

🎯 Common Workflows

Database Exploration

# 1. List all schemas
schemas = list_schemas()

# 2. List tables in schema
tables = list_objects(schema_name="public", object_type="table")

# 3. Get table details
details = get_object_details(
    schema_name="public",
    object_name="users",
    object_type="table"
)

Performance Optimization

# 1. Identify slow queries
slow_queries = get_top_queries(sort_by="mean_time", limit=10)

# 2. Analyze specific query
plan = explain_query(sql="SELECT * FROM orders WHERE...", analyze=True)

# 3. Get index recommendations
recommendations = analyze_query_indexes(queries=[...])

# 4. Check database health
health = analyze_db_health(health_type="all")

Secure Data Access

# Always use parameter binding
result = execute_sql(
    sql="SELECT * FROM users WHERE email = %s AND role = %s",
    params=["user@example.com", "admin"]
)

🔒 Security Best Practices

  1. Always Use Parameter Binding

    # ✅ SECURE
execute_sql("SELECT * FROM users WHERE id = %s", params=[123])

# ❌ VULNERABLE
execute_sql(f"SELECT * FROM users WHERE id = {user_id}")

  2. Use Restricted Mode in Production

    • Enables read-only operations
    • Advanced query validation
    • Resource limits

  3. Monitor Query Performance

    • Regular get_top_queries() checks
    • Watch for unusual patterns
    • Set up alerts for slow queries

📚 Related Documentation

See Home for more tool categories.

Clone this wiki locally