0
Execute arbitrary Query
One script reply has been approved by the moderators Verified

Execute an arbitrary query on a PostgreSQL resource, more at: https://www.windmill.dev/docs/getting_started/scripts_quickstart/sql

Created by henri186 48 days ago Viewed 4295 times
0
Submitted by henri186 Python3
Verified 48 days ago
1
from typing import TypedDict, Dict, Any
2
import psycopg2
3

4
# Define the PostgreSQL resource type as specified
5
class postgresql(TypedDict):
6
    host: str
7
    port: int
8
    user: str
9
    dbname: str
10
    sslmode: str
11
    password: str
12
    root_certificate_pem: str
13

14
def main(query: str, db_config: postgresql) -> Dict[str, Any]:
15
    # Connect to the PostgreSQL database
16
    conn = psycopg2.connect(
17
        host=db_config["host"],
18
        port=db_config["port"],
19
        user=db_config["user"],
20
        password=db_config["password"],
21
        dbname=db_config["dbname"],
22
        sslmode=db_config["sslmode"],
23
        sslrootcert=db_config["root_certificate_pem"],
24
    )
25

26
    # Create a cursor object
27
    cur = conn.cursor()
28

29
    # Execute the query
30
    cur.execute(query)
31

32
    # Fetch all rows from the last executed statement
33
    rows = cur.fetchall()
34

35
    # Close the cursor and connection
36
    cur.close()
37
    conn.close()
38

39
    # Convert the rows to a list of dictionaries to make it more readable
40
    columns = [desc[0] for desc in cur.description]
41
    result = [dict(zip(columns, row)) for row in rows]
42

43
    return result