aurora_query_check.py - Simple CLI tool to check if SELECT queries are hitting the Aurora writer. Helps debug read/write routing issues quickly.

README.md

Aurora Query Routing Checker

A simple command-line diagnostic tool that helps you identify whether SELECT queries are hitting the writer instance in your Amazon Aurora PostgreSQL cluster. Use this tool to quickly confirm routing problems and take action.

What You See When You Run the Tool

==============================
 Aurora Query Routing Checker
==============================

1) Check instance role
2) Show SELECT queries
3) Exit

Select an option: _

Example output from Option 2:

----------------------------------------
Aurora SELECT Query Report
Saved to: aurora_session_log.txt
Timestamp: 2025-06-27 14:42:01 UTC
----------------------------------------

[Instance Role: WRITER]

1. app_user     10.1.4.28     SELECT id, name FROM produ...
2. metrics_bot  10.1.7.33     SELECT count(*) FROM logs ...

(2 queries shown — limited view)

Note: SELECTs are hitting the WRITER. Consider routing them to read replicas.

How to Set Up and Run the Tool

Step 1: Create a new folder and enter it

mkdir aurora_diag
cd aurora_diag

Step 2: Create a Python virtual environment

python3 -m venv venv

Step 3: Activate the virtual environment

source venv/bin/activate

Step 4: Install required package

pip install psycopg2-binary

Step 5: Paste the script into a file

Create a file named aurora_query_check.py and paste in the full script contents from the Gist.

nano aurora_query_check.py

Paste, save, and exit.

Step 6: Set your environment variables

export DB_HOST=your-db-host.rds.amazonaws.com
export DB_PORT=5432
export DB_NAME=your_db_name
export DB_USER=your_db_user
export DB_PASS=your_password

Step 7: Run the script

python aurora_query_check.py

Step 8: Deactivate when finished

deactivate

Log output is saved to aurora_session_log.txt in the same folder.

---

This tool was designed to help engineers quickly detect routing problems in Aurora PostgreSQL clusters—especially in environments where read/write traffic separation is unclear or inconsistent. If you find SELECTs going to your writer, you’ll want to review query endpoints, app logic, and connection pooling strategies.

aurora_query_check.py

import os
import psycopg2


def get_connection():
    try:
        conn = psycopg2.connect(
            host=os.getenv("DB_HOST"),
            port=os.getenv("DB_PORT", 5432),
            dbname=os.getenv("DB_NAME"),
            user=os.getenv("DB_USER"),
            password=os.getenv("DB_PASS")
        )
        return conn
    except Exception as e:
        print(f"Connection failed: {e}")
        return None


def check_instance_role(conn):
    with conn.cursor() as cur:
        cur.execute("SELECT pg_is_in_recovery();")
        result = cur.fetchone()
        role = "READER" if result[0] else "WRITER"
        print(f"\nThis instance is: {role}\n")


def show_select_queries(conn):
    with conn.cursor() as cur:
        cur.execute("""
            SELECT usename, client_addr, left(query, 40)
            FROM pg_stat_activity
            WHERE state = 'active'
              AND query ~* '^\\s*select'
              AND pid <> pg_backend_pid();
        """)
        rows = cur.fetchall()
        print("\n-------------------------------")
        print("  SELECT Queries on This Node")
        print("-------------------------------\n")
        cur.execute("SELECT pg_is_in_recovery();")
        is_reader = cur.fetchone()[0]
        print(f"[Instance Role: {'READER' if is_reader else 'WRITER'}]\n")
        if not rows:
            print("No active SELECT queries found.\n")
        else:
            for idx, row in enumerate(rows, start=1):
                print(f"{idx}. {row[0]:<15} {row[1]:<15} {row[2]}...")
            print(f"\n({len(rows)} queries shown — limited view)")
            if not is_reader:
                print("\nNote: SELECTs are hitting the WRITER. Consider routing them to read replicas.\n")


def main():
    print("==============================")
    print(" Aurora Query Routing Checker")
    print("==============================\n")
    conn = get_connection()
    if not conn:
        return

    try:
        while True:
            print("1) Check instance role")
            print("2) Show SELECT queries")
            print("3) Exit\n")
            choice = input("Select an option: ")

            if choice == '1':
                check_instance_role(conn)
            elif choice == '2':
                show_select_queries(conn)
            elif choice == '3':
                print("Exiting...\n")
                break
            else:
                print("Invalid option. Please try again.\n")
    finally:
        conn.close()


if __name__ == "__main__":
    main()