Edits history of script submission #6003 for ' Execute arbitrary Query (postgresql)'

• python3

  Ap­pro­ved version

  Edit in Windmill

  Embed

  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 henri186 61 days ago