ae.db_core

database connection and data manipulation base classes

This module is providing generic base classes and helper methods to implement and integrate any database driver that is compatible to the Python DB API. The base classes are providing already all attributes and methods to handle database connections, data selections and manipulations.

Additionally this package is defining generalized interfaces to switch dynamically between databases without the need to adapt the code of your application - simply by selecting another database connector class within your configuration files.

basic usage

The abstract base class DbBase allows you to implement a db-specific class with few lines of code. Simply inherit from DbBase and implement a connect() method that is setting the conn instance attribute to a Python DB API compatible connection object.

For a fictive specific database driver package SpecificDbDriver that is providing a connect method, a minimal example would look like:

from ae.db_core import DbBase
...
class SpecificDb(DbBase):
    def connect(self) -> str:
        self.last_err_msg = ''
        try:
           self.conn = SpecificDbDriver.connect(**self.connect_params())
        except Exception as ex:
            self.last_err_msg = f"SpecificDb-connect() error: {ex} for {self}"
        else:
            self.create_cursor()
       return self.last_err_msg

The helper method connect_params() provided by this module is helping you to merge any specified database system credentials and features into a dict to be passed as connection parameters to the database-specific connection class/function/factory of the database driver.

Another helper create_cursor() provided by DbBase can be used in the connect() method to create a Python DB API-compatible cursor object and assign it to the curs attribute.

In this implementation your class SpecificDb does automatically also inherit useful methods from DbBase to execute INSERT, DELETE, UPDATE and UPSERT commands, to run SELECT queries and to call stored procedures.

Hint

Examples of an implementation of a specific database class are e.g. the portion packages ae.db_pg for Postgres and ae.db_ora for Oracle.

connection parameters

DbBase supports the following connection parameter keys:

user : user name or database account name.

password : user account password.

dbname : name of the database to connect.

host : host name or IP address of the database server.

port: port number of the database server.

Some database drivers are using default values for most of these connection parameters, if they are not specified. The mandatory connection parameters can differ for different database drivers. Most of them need at least a user name and password.

Hint

Most database drivers like e.g. psycopg2 or pg8000 for Postgres databases are supporting all of these connection parameter keys. For others like e.g. cx_Oracle which are not supporting some of them (like e.g. host and dbname) you may also need to overwrite the connect_params() method to convert/adopt unsupported connection parameter keys.

Alternatively you can provide the connection parameters needed by the database driver connector with a single database url string in the SQLAlchemy Database URL format:

dialect+driver://user:password@host:port/dbname

So if your connection parameters dict - respective your database system credentials and features - providing a url key then connect_params() will parse and split the url into separate connection parameters.

Note

When you specify a connection parameter as separate key and also within the url value then the separate key value will have preference. The scheme part (dialect+driver) including the following colon character of the URL string are not evaluated and can be omitted.

bind variables format

Bind variables in your sql query strings have to be formatted in the named parameter style. If your database driver is only supporting the pyformat parameter style, then overload the __init__ method and set the param_style to 'pyformat' and all sql query strings will be automatically converted by DbBase before they get sent to the database:

class SpecificDb(DbBase):
    def __init__(self, ...)
        super().__init__(...)
        self.param_style = 'pyformat'
    ...

Module Attributes

`NAMED_BIND_VAR_PREFIX`	character to mark a bind variable in a sql query
`CHK_BIND_VAR_PREFIX`	bind variable name prefix, to allow for same column/-name a new/separate value in e.g.

Functions

connect_args_from_params(conn_params)

split dict with database connection parameters into credentials and features.

Classes

DbBase(system)

abstract database connector base class for the ae namespace database system layers.

NAMED_BIND_VAR_PREFIX: str = ':': character to mark a bind variable in a sql query

CHK_BIND_VAR_PREFIX: str = 'CV_': bind variable name prefix, to allow for same column/-name a new/separate value in e.g. the SET clause and an old value in the WHERE clause. Gets added to bind variables in filters/chk_values and extra where clauses.

connect_args_from_params(conn_params)[source]

split dict with database connection parameters into credentials and features.

Parameters:: conn_params¶ (Dict[str, Any]) – connection params dict.
Return type:: Tuple[Dict[str, str], List[str]]
Returns:: tuple of credentials dict and features list.

_normalize_col_values(col_values)[source]

convert empty strings into real None values.

Parameters:: col_values¶ (Dict[str, Any]) – dict of column values (dict key is the column name).
Return type:: Dict[str, Any]
Returns:: col_values dict where empty values got replaced with None (also in original dict).

_prepare_in_clause(sql, bind_vars=None, additional_col_values=None)[source]

replace list bind variables used in an IN clause with separate bind variables for each list item.

Parameters:

sql¶ (str) – query to be executed.
bind_vars¶ (Optional[Dict[str, Any]]) – dict of all available bind variables for the execution.
additional_col_values¶ (Optional[Dict[str, Any]]) – additional bind variables.

Return type:

Tuple[str, Dict[str, Any]]

Returns:

tuple of adapted query string and joined bind variables.

_rebind(chk_values=None, where_group_order='', bind_vars=None, extra_bind=None)[source]

merge where_group_order string with chk_values filter dict and merge/rename bind variables.

Parameters:

chk_values¶ (Optional[Dict[str, Any]]) –
dict with column_name: value items, used to filter/restrict the resulting rows.

This method compiles this dict into a sql WHERE clause expression. If you also passed additional sql clauses into the where_group_order then it will be merged with the compiled expression. The names of the sql parameters and related bind variables are build from the column names/keys of this dict, and will be prefixed with CHK_BIND_VAR_PREFIX.

Passing None or a empty dict to this and to the extra_bind arguments will disable any filtering. If only this argument is None/empty then the first item of the extra_bind dict will be used as filter.
where_group_order¶ (str) – sql part with optional WHERE/GROUP/ORDER clauses (the part after the WHERE), including bind variables in the named parameter style.
bind_vars¶ (Optional[Dict[str, Any]]) – dict with bind variables (variable name has to be prefixed in where_group_order with CHK_BIND_VAR_PREFIX).
extra_bind¶ (Optional[Dict[str, Any]]) – additional dict with bind variables (variable name has NOT to be prefixed/adapted in where_group_order argument).

Return type:

Tuple[Optional[Dict[str, Any]], str, Dict[str, Any]]

Returns:

tuple of corrected/rebound values of chk_values, where_group_order and bind_vars.

class DbBase(system)[source]

Bases: SystemConnectorBase, ABC

abstract database connector base class for the ae namespace database system layers.

This class inherits from SystemConnectorBase the following attributes:

system: instance of the related ae-sys_core.SystemBase class.

console_app: instance of the application using this database system.

last_err_msg: the error message string of the last error or an empty string if no error occurred in the last database action/operation.

__init__(system)[source]

create instance of generic database object (base class for real database like e.g. postgres or oracle).

Parameters:

system¶ (SystemBase) –

SystemBase instance. Providing useful attributes, like e.g.:

credentials: dict with database driver specific account credentials.

features: list of features. Features will also be passed as connection parameters to the database driver. The main differences to credentials are that a feature without any value will be interpreted as boolean True and that features can have any value that can be specified as a literal (like int or even datetime).

console_app: instance of the application using this database system.

conn: database driver connection

curs: database driver cursor

param_style: str: database driver bind variable/parameter style

abstract connect()[source]

sub-class has to implement this connect method

Return type:: str

_adapt_sql(sql, bind_vars)[source]

replace the parameter style of bind variables from pyformat into named.

Parameters:

sql¶ (str) – query to scan for named bind variables.
bind_vars¶ (Dict[str, Any]) – dict of all available bind variables.

Return type:

str

Returns:

adapted query string.

Note

For database drivers - like psycopg2 - that are support only the pyformat parameter style syntax (in the format %(bind_var)s) the sql query string will be adapted, by converting all bind variables from the parameter style named into pyformat.

The returned query will be unchanged for all other database drivers (that are directly supporting the named parameter style).

call_proc(proc_name, proc_args, ret_dict=None)[source]

execute stored procedure on database server.

Parameters:

proc_name¶ (str) – name of the stored procedure.
proc_args¶ (Sequence) – tuple of parameters/arguments passed to the stored procedure.
ret_dict¶ (Optional[Dict[str, Any]]) – optional dict - if passed then the dict item with the key return will be set/updated to the value returned from the stored procedure/database.

Return type:

str

Returns:

empty string if no error occurred else the error message.

close(commit=True)[source]

close the connection to the database driver and server.

Parameters:: commit¶ (bool) – pass False to prevent commit (and also execute rollback) before closure of connection.
Return type:: str
Returns:: empty string if no error occurred else the error message.

connect_params()[source]

merges self.system.credentials with self.system.features into a database driver connection parameters dict.

Return type:

Dict[str, Any]

Returns:

new dict with the items from self.system.credentials and then extended with the entries of self.system.features.

Credential values are always of type str, only features can be of any type. Features without a specified value are set True. All keys of the returned dict will be in lower-case.

create_cursor()[source]: allow sub-class to create Python DB API-conform database driver cursor

cursor_description()[source]

return description text if opened cursor or None if cursor is closed or not yet opened.

Return type:: Optional[str]

fetch_all()[source]

fetch all the rows found from the last executed SELECT query.

Return type:: Sequence[Tuple]
Returns:: empty list on error or if query result is empty, else a list of database rows.

fetch_value(col_idx=0)[source]

fetch the value of a column of the first/next row of the found rows of the last SELECT query.

Parameters:: col_idx¶ (int) – index of the column with the value to fetch and return.
Return type:: Any
Returns:: value of the column at index col_idx.

execute_sql(sql, commit=False, bind_vars=None)[source]

execute sql query with optional bind variables.

Parameters:

sql¶ (str) – sql query to execute.
commit¶ (bool) – pass True to execute a COMMIT command directly after the query execution.
bind_vars¶ (Optional[Dict[str, Any]]) – optional dict with bind variables.

Return type:

str

Returns:

empty string if no error occurred else the error message.

delete(table_name, chk_values=None, where_group_order='', bind_vars=None, commit=False)[source]

execute a DELETE command against a table.

Parameters:

table_name¶ (str) – name of the database table.
chk_values¶ (Optional[Dict[str, Any]]) – dict of column names/values to identify the record(s) to delete.
where_group_order¶ (str) – extra sql added after the WHERE clause (merged with chk_values by _rebind()). This string can include additional WHERE expressions with extra bind variables.
bind_vars¶ (Optional[Dict[str, Any]]) – dict of extra bind variables (key=variable name, value=value).
commit¶ (bool) – bool value to specify if commit should be done. Pass True to commit.

Return type:

str

Returns:

last error message or empty string if no errors occurred.

insert(table_name, col_values, returning_column='', commit=False)[source]

execute an INSERT command to add one record to a database table.

Parameters:

table_name¶ (str) – name of the database table.
col_values¶ (Dict[str, Any]) – dict of inserted column values with the column name as key.
returning_column¶ (str) – name of column which value will be returned by next fetch_all/fetch_value() call.
commit¶ (bool) – bool value to specify if commit should be done. Pass True to commit.

Return type:

str

Returns:

last error message or empty string if no errors occurred.

select(from_join='', cols=(), chk_values=None, where_group_order='', bind_vars=None, hints='')[source]

execute a SELECT query against a database table.

Parameters:

from_join¶ (str) – name(s) of the involved database table(s), optional with JOIN clause(s). Passing an empty string results in a SELECT statement without the FROM keyword.
cols¶ (Sequence[str]) – sequence of the column names that will be selected and included in the resulting data-rows.
chk_values¶ (Optional[Dict[str, Any]]) – dict of column names/values to identify selected record(s).
where_group_order¶ (str) – extra sql added after the WHERE clause (merged with chk_values by _rebind()). This string can include additional WHERE expressions with extra bind variables, ORDER BY and GROUP BY expressions. These special/extra bind variables have to be specified in the bind_vars argument and have to be prefixed with the string ‘CV_’ in the WHERE clause (see also the CHK_BIND_VAR_PREFIX data constant in this module).
bind_vars¶ (Optional[Dict[str, Any]]) – dict of extra bind variables (key=variable name, value=value).
hints¶ (str) – optional SELECT optimization hint string.

Return type:

str

Returns:

last error message or empty string if no errors occurred. Use the methods fetch_all() or fetch_value() to retrieve the resulting data-rows.

update(table_name, col_values, chk_values=None, where_group_order='', bind_vars=None, commit=False, locked_cols=())[source]

execute an UPDATE command against a database table.

Parameters:

table_name¶ (str) – name of the database table.
col_values¶ (Dict[str, Any]) – dict of inserted/updated column values with the column name as key.
chk_values¶ (Optional[Dict[str, Any]]) – dict of column names/values to identify affected record(s). If not passed then the first name/value of col_values is used as primary key check/filter value.
where_group_order¶ (str) – extra sql added after the WHERE clause (merged with chk_values by _rebind()). This string can include additional WHERE expressions with extra bind variables, ORDER BY and GROUP BY expressions. These special/extra bind variables have to be specified in the bind_vars argument and have to be prefixed with the string ‘CV_’ (see also the CHK_BIND_VAR_PREFIX data constant in this module).
bind_vars¶ (Optional[Dict[str, Any]]) – dict of extra bind variables (key=variable name, value=value).
commit¶ (bool) – bool value to specify if commit should be done. Pass True to commit.
locked_cols¶ (Sequence[str]) – list of column names not be overwritten on update of column value is not empty.

Return type:

str

Returns:

last error message or empty string if no errors occurred.

upsert(table_name, col_values, chk_values, where_group_order='', bind_vars=None, returning_column='', commit=False, locked_cols=(), multiple_row_update=True)[source]

execute an INSERT or UPDATE command against a record of a database table (UPDATE if record already exists).

Parameters:

table_name¶ (str) – name of the database table.
col_values¶ (Dict[str, Any]) – dict of inserted/updated column values with the column name as key.
chk_values¶ (Dict[str, Any]) – dict of column names/values to identify affected record(s), also used to check if record already exists (and data has to updated instead of inserted). If not passed then the first name/value of col_values is used as primary key check/filter value.
where_group_order¶ (str) – extra sql added after the WHERE clause (merged with chk_values by _rebind()). This string can include additional WHERE expressions with extra bind variables, ORDER BY and GROUP BY expressions. These special/extra bind variables have to be specified in the bind_vars argument and have to be prefixed with the string ‘CV_’ in the query string (see also the CHK_BIND_VAR_PREFIX data constant in this module).
bind_vars¶ (Optional[Dict[str, Any]]) – dict of extra bind variables (key=variable name, value=value).
returning_column¶ (str) – name of column which value will be returned by next fetch_all/fetch_value() call.
commit¶ (bool) – bool value to specify if commit should be done. Pass True to commit record changes.
locked_cols¶ (Sequence[str]) – list of column names not be overwritten on update of column value is not empty.
multiple_row_update¶ (bool) – allow update of multiple records with the same chk_values.

Return type:

str

Returns:

last error message or empty string if no errors occurred.

commit(reset_last_err_msg=False)[source]

commit the current transaction if the database driver supports/implements a commit method.

Parameters:: reset_last_err_msg¶ (bool) – pass True to reset the last error message of this instance before the commit.
Return type:: str
Returns:: last error message or empty string if no error happened.

rollback(reset_last_err_msg=False)[source]

roll the current transaction back if the DB driver supports transactions and does have a rollback method.

Parameters:: reset_last_err_msg¶ (bool) – pass True to reset the last error message of this instance before the rollback.
Return type:: str
Returns:: last error message or empty string if no error happened.

get_row_count()[source]

determine rowcount of last executed query.

Return type:: int
Returns:: the number of affected rows of the last query.

selected_column_names()[source]

determine the column names fo the last executed SELECT query.

Return type:: List[str]
Returns:: list of column names - order by column index.

static thread_lock_init(table_name, chk_values)[source]

created named locking instance for passed table and filter/check expression.

Return type:: NamedLocks
Returns:: NamedLocks instance for this table and filter.