ConnectorTypeSQLPG

class connector_types.connector_type_sqlpg.ConnectorTypeSQLPG

Interact with a PostgreSQL database. This connector type supports the execute, fetch, fetchrow, and fetchval commands. Each command expects an SQL query and returns the status, list, record or field value respectively.

Consult the PostgreSQL SQL language documentation at https://www.postgresql.org/docs/12/tutorial-sql.html for more information.

Inputs

Name	Type	Default	Description
check_hostname	bool	True	If set, the hostname of the PostgreSQL server is checked against the server_ca certificate
client_cert	str	None
client_key	str	None
database	str	pg_catalog
execute	str	None
executemany	list	None
fetch	str	None
fetchrow	str	None
fetchval	str	None
host	str
params	list	None
password	str	None
port	int	5432
server_ca	str	None
temp_path	str	/c/tmp
transaction	list	None
user	str	postgres

Outputs

Name	Type	Default	Description
execution_id	int		The ID of the connection execution
message	str		The ended message for the connection. If the connection ended with an error, the message will contain information about what went wrong
result	object
status	str		The ended status for the connection. Either "success" or "error".

Constants

input_list = ['check_hostname', 'client_cert', 'client_key', 'database', 'execute', 'executemany', 'fetch', 'fetchrow', 'fetchval', 'host', 'params', 'password', 'port', 'server_ca', 'temp_path', 'transaction', 'user'] output_list = ['result'] ssl_context_inputs = ['check_hostname', 'client_cert', 'client_key', 'server_ca'] version = 1

Methods

execute ()

log (message)

one_of_inputs (options)

run ()

Example

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    postgres_server_version = this.connect(
        connector_type='SQLPG',
        host='my-postgres-server',
        fetchval='SELECT version()',
    ).get('output_value')['result']
    this.log(postgres_server_version=postgres_server_version)

    postgres_server_events = this.connect(
        connector_type='SQLPG',
        host='my-postgres-server',
        database='testlocal',
        execute='CREATE TABLE IF NOT EXISTS mytab (a int, b int);',
    ).get('output_value')['result']
    this.log(postgres_server_events=postgres_server_events)

    postgres_server_events = this.connect(
        connector_type='SQLPG',
        host='my-postgres-server',
        database='testlocal',
        transaction=[['INSERT INTO mytab (a, b) VALUES ($1, $2);', (1,2)],
                    ['INSERT INTO mytab (a, b) VALUES ($1, $2);', (3,4)],
                    ['INSERT INTO mytab (a, b) VALUES ($1, $2);', (6,7)]
                    ],
    ).get('output_value')['result']
    this.log(postgres_server_events=postgres_server_events)

    postgres_server_events = this.connect(
        connector_type='SQLPG',
        host='my-postgres-server',
        database='testlocal',
        executemany=['INSERT INTO mytab (a, b) VALUES ($1, $2);',
                    [(1,2), (3,4), (6,7)]],
    ).get('output_value')['result']
    this.log(postgres_server_events=postgres_server_events)

    return this.success('all done')