ae.db_pg
postgres database layer
this module provides the class PostgresDb
to connect and interact with a Postgres database server.
the communication with the database driver is done with great help of the psycopg2 pypi package.
basic usage of ae.db_pg
to create an instance of the class PostgresDb
you first have to create a SystemBase
instance. this can be done either programmatically by providing an application instance (of the class
ConsoleApp
or an inherited subclass of it) plus any database parameters, like required
credentials and any database configuration features/options:
app = ConsoleApp()
system = SystemBase('system-id', app, dict(User='user name', Password='password`, Dbname=...), ...)
alternatively provide all system-specific info within the ae config files and let
UsedSystems
load it:
system = used_systems['system-id']
finally pass the database parameters in system to create an instance of PostgresDb
:
pg_db = PostgresDb(system)
then call the connect()
method of this instance to connect to the Postgres database server:
error_message = pg_db.connect()
if error_message:
print(error_message)
if the connection could not be established then connect()
is returning an error message string. if
the return value is an empty string then you can use all the methods provided by DbBase
, like e.g.
update()
:
error_message = pg_db.update('my_table', {'my_col': 'new value'})
if error_message:
print(error_message)
error_message = pg_db.rollback()
an explicit call of rollback()
is only needed if you use transactions (autocommit is False).
in this case you should also use commit()
at the end of each transaction to store any data
updates:
error_message = pg_db.commit()
alternatively you can use the commit argument that is provided by the DbBase
DML methods: by
passing a True value to this argument, the method will automatically execute a commit()
call
for you if no error occurred in the DML method:
error_message = pg_db.update('table`, {'column': 369}, commit=True)
finally after all database actions are done you can close the connection to the databases server with the
close()
method:
error_message = pg_db.close()
Classes
|
an instance of this class represents a Postgres database. |
- class PostgresDb(system)[source]
Bases:
DbBase
an instance of this class represents a Postgres database.
- __init__(system)[source]
create instance of Postgres database object
- Parameters:
system¶ (
SystemBase
) – instance of aSystemBase
class.
SystemBase
(defined in the moduleae.sys_core
) is providing the credentials and features, which get retrieved from config files, then converted byconnect_params()
into connection parameters to connect to the Postgres database.for connections via SSL to the Postgres server you have to add either the connection parameters sslmode, sslcert and sslkey or sslrootcert and sslcrl (depending on the configuration of your server).
features : optional list of features.
- connect()[source]
connect this instance to the Postgres database server, using the credentials provided at instantiation.
- Return type:
- Returns:
error message in case of error or empty string if not.
- execute_sql(sql, commit=False, bind_vars=None, auto_commit=False)[source]
execute sql query or sql command.
- Parameters:
- Return type:
- Returns:
last error message or empty string if no errors occurred.
Hint
overwriting generic execute_sql for Postgres because if auto_commit is False then a db error is invalidating the connection until it gets rolled back (optionally to a save-point). unfortunately psycopg2 does not provide/implement save-points. could be done alternatively with execute(“SAVEPOINT NonAutoCommErrRollback”) but RELEASE/ROLLBACK makes it complicated (see also https://stackoverflow.com/questions/2370328/continuing-a-transaction-after-primary-key-violation-error):
save_point = None if auto_commit else self.conn.setSavepoint('NonAutoCommErrRollback') super().execute_sql(sql, commit=commit, auto_commit=auto_commit, bind_vars=bind_vars) if save_point: if self.last_err_msg: self.conn.rollback(save_point) else: self.conn.releaseSavepoint(save_point) return self.last_err_msg
therefore KISS - a simple rollback will do it also.