`config_from_pg.db`

Database helper module. Modified from https://gist.github.com/rustprooflabs/3b8564a8e7b7fe611436b30a95b7cd17, adapted to psycopg 3 from psycopg2.

Module Contents

`prepare`()	Ensures latest pgconfig.settings view exists in DB to generate config.
`ensure_schema_exists`()	Ensures the pgconfig schema exists.
`ensure_view_exists`()	Ensures the view pgconfig.settings exists.
`select_one`(→ dict)	Runs SELECT query that will return zero or 1 rows.
`select_multi`(→ list)	Runs SELECT query that will return multiple. params is optional.
`get_db_string`(→ str)	Prompts user for details to create connection string
`get_db_conn`()	Uses DB_STRING to establish psycopg connection.
`_execute_query`(sql_raw, params, qry_type)	Handles executing all types of queries based on the qry_type passed in.

config_from_pg.db.prepare()[source]: Ensures latest pgconfig.settings view exists in DB to generate config.

config_from_pg.db.ensure_schema_exists()[source]: Ensures the pgconfig schema exists.

config_from_pg.db.ensure_view_exists()[source]: Ensures the view pgconfig.settings exists.

config_from_pg.db.select_one(sql_raw: str, params: dict) → dict[source]

Runs SELECT query that will return zero or 1 rows.

Parameters:

sql_raw (str) –
params (dict) – Params is required, can be None if query returns a single row such as SELECT version();

Returns:

data

Return type:

dict

config_from_pg.db.select_multi(sql_raw, params=None) → list[source]

Runs SELECT query that will return multiple. params is optional.

Parameters:

Returns:

data – List of dictionaries.

Return type:

list

config_from_pg.db.get_db_string() → str[source]

Prompts user for details to create connection string

config_from_pg.db.get_db_conn()[source]: Uses DB_STRING to establish psycopg connection.

config_from_pg.db._execute_query(sql_raw, params, qry_type)[source]

Handles executing all types of queries based on the qry_type passed in. Returns False if there are errors during connection or execution.

if results == False:
print(‘Database error’)

else:
print(results)

You cannot use if not results: b/c 0 results is a false negative.