from typing import TypedDict, Dict, Any
import psycopg2
# Define the PostgreSQL resource type as specified
class postgresql(TypedDict):
host: str
port: int
user: str
dbname: str
sslmode: str
password: str
root_certificate_pem: str
def main(query: str, db_config: postgresql) -> Dict[str, Any]:
# Connect to the PostgreSQL database
conn = psycopg2.connect(
host=db_config["host"],
port=db_config["port"],
user=db_config["user"],
password=db_config["password"],
dbname=db_config["dbname"],
sslmode=db_config["sslmode"],
sslrootcert=db_config["root_certificate_pem"],
)
# Create a cursor object
cur = conn.cursor()
# Execute the query
cur.execute(query)
# Fetch all rows from the last executed statement
rows = cur.fetchall()
# Close the cursor and connection
cur.close()
conn.close()
# Convert the rows to a list of dictionaries to make it more readable
columns = [desc[0] for desc in cur.description]
result = [dict(zip(columns, row)) for row in rows]
return result
Submitted by hugo697 285 days ago