types-psycopg2

import psycopg2 import os # Configure connection details from environment variables or provide defaults dbname = os.environ.get('PG_DBNAME', 'testdb') user = os.environ.get('PG_USER', 'postgres') password = os.environ.get('PG_PASSWORD', 'password') host = os.environ.get('PG_HOST', 'localhost') port = os.environ.get('PG_PORT', '5432') try: # Establish a database connection conn = psycopg2.connect(f"dbname={dbname} user={user} password={password} host={host} port={port}") conn.autocommit = True # For simplicity in this quickstart # Open a cursor to perform database operations with conn.cursor() as cur: # Create a table cur.execute("CREATE TABLE IF NOT EXISTS test_table (id SERIAL PRIMARY KEY, name VARCHAR(255), age INTEGER)") print("Table 'test_table' ensured to exist.") # Insert data cur.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ('Alice', 30)) cur.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ('Bob', 24)) print("Data inserted.") # Query data cur.execute("SELECT id, name, age FROM test_table ORDER BY id") rows = cur.fetchall() print("Fetched data:") for row in rows: print(f" ID: {row[0]}, Name: {row[1]}, Age: {row[2]}") print("Quickstart example completed successfully.") except psycopg2.Error as e: print(f"Database error: {e}") if 'conn' in locals() and not conn.closed: conn.rollback() finally: if 'conn' in locals() and not conn.closed: conn.close() print("Database connection closed.")