pgspecial

import os import psycopg2 from pgspecial.main import PGSpecial try: # Connection details from environment variables for security/flexibility DB_NAME = os.environ.get('PG_DB_NAME', 'postgres') DB_USER = os.environ.get('PG_DB_USER', 'postgres') DB_PASSWORD = os.environ.get('PG_DB_PASSWORD', '') DB_HOST = os.environ.get('PG_DB_HOST', 'localhost') DB_PORT = os.environ.get('PG_DB_PORT', '5432') conn = psycopg2.connect( dbname=DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT ) cur = conn.cursor() pgspecial = PGSpecial() # Example: List tables in the current database print("\n--- Executing \\dt ---") for title, rows, headers, status in pgspecial.execute(cur, '\\dt'): if title: print(f"Title: {title}") if headers: print(f"Headers: {', '.join(headers)}") if rows: for row in rows: print(f"Row: {row}") if status: print(f"Status: {status}") # Example: Describe a specific table (if one exists, e.g., 'your_table_name') # You might need to create a dummy table for this to show results try: cur.execute("CREATE TABLE IF NOT EXISTS pgspecial_example (id SERIAL PRIMARY KEY, name VARCHAR(50))") conn.commit() print("\n--- Executing \\d pgspecial_example ---") for title, rows, headers, status in pgspecial.execute(cur, '\\d pgspecial_example'): if title: print(f"Title: {title}") if headers: print(f"Headers: {', '.join(headers)}") if rows: for row in rows: print(f"Row: {row}") if status: print(f"Status: {status}") except psycopg2.Error as e: print(f"Could not create/describe example table: {e}") except psycopg2.Error as e: print(f"Error connecting to PostgreSQL: {e}") print("Please ensure PostgreSQL is running and connection details (PG_DB_NAME, PG_DB_USER, PG_DB_PASSWORD, PG_DB_HOST, PG_DB_PORT) are correct.") finally: if 'conn' in locals() and conn: conn.close()