Base Table classes

Implement some more abstract base model (table) classes.

These classes add some more generic database functionalities:

Some generic methods which run full text SQL statements.

A generic method to run stored procedures which return data.

A generic method which takes a list of records to be inserted, and records to be updated, and writes them to the target database table in a single call.

Transaction atomicity. Multiple database operations involving several different tables can be wrapped under a single transaction, so that all can be committed or rolled back as appropriate.

Applications should subclass their own tables from either one of these two classes ReadOnlyTable and WriteCapableTable.

Please note, WriteCapableTable requires the target database to implement unique_id table and associated get_unique_id stored method, please see Database requirements for more detail.

Relevant test modules:

./tests/test_25_base_table_crud_methods_postgresql.py

./tests/test_26_base_table_crud_methods_mysql.py

./tests/test_30_base_table_exception_postgresql.py

./tests/test_31_base_table_exception_mysql.py

class bh_database.base_table.BaseTable(**kwargs)

Bases: BaseSQLAlchemy

An abstract base model (table).

Parameters:: kwargs – column name, value pairs.

For example:

employees = Employees(emp_no=456000, ..., hire_date='2021-11-02')

in:

class Employees(WriteCapableTable):
    emp_no = Column(Integer, primary_key=True)
    ...
    hire_date = Column(Date, nullable=False)

employees = Employees(emp_no=456000, ..., hire_date='2021-11-02')

assert employees.emp_no == 456000
...
assert employees.hire_date == '2021-11-02' 

where WriteCapableTable is an indirect subclass of BaseTable.

In general, for this abstract class and its abstract descendant classes, do not access query when not assigned. E.g.:

assert BaseTable.query == None

It results in the following exception:

ArgumentError("Column expression, FROM clause, or other columns clause element expected, <class 'bh_database.core.BaseSQLAlchemy'>.")

A relevant test module ./tests/test_17_base_table_methods.py.

as_dict() → dict

Convert all column-value pairs of model instance to a dictionary.

For a full example usage, please see test module ./tests/test_17_base_table_methods.py.

References:

Returns:

all column-value pairs as a dictionary.

Return type:

dict.

class bh_database.base_table.ReadOnlyTable(**kwargs)

Bases: BaseTable

Implement a read-only abstract base model (table) class.

This table class should be the parent class for lookup tables, whom the contents seldom change.

Read-only in the sense that this class has only a public method to query database. It is only a semantic classification. It is also an indirect child of BaseSQLAlchemy class, it can use class attributes session and query to alter data directly.

In general, for this abstract class and its abstract descendant classes, do not access query when not assigned. E.g.:

assert ReadOnlyTable.query == None

It results in the following exception:

ArgumentError("Column expression, FROM clause, or other columns clause element expected, <class 'bh_database.core.BaseSQLAlchemy'>.")

run_select_sql(sql: str, auto_session=False) → ResultStatus

Run a SELECT SQL full text statement and returns the result.

It is assumed a SELECT SQL statement, there is no check enforced.

Parameters:

sql (str) – the full text SELECT SQL statement.
auto_session (bool) – upon an operation on the underlying database, SQLAlchemy auto starts a transaction if there is not one in progress. If this method is in a single call, the caller should set this param to True to get rid of the transaction when finished: if the call was successful, the transaction is committed, otherwise it rolls back. If it is called as a part of an ongoing transaction, in which case SQLAlchemy does not start another transaction, then just ignore this param, the caller is responsible for managing transaction atomicity.

Returns:

ResultStatus.

Further illustrations of return value, as a dictionary.

On successful:

{
    "status": {
        "code": 200,
        "text": "Data has been retrieved successfully."                    
    },
    "data": [
        {...},
        ...,
        {...}
    ]
}

where 200 is HTTPStatus.OK.value. 200 does not mean the SELECT SQL statement results in any data retrieved.

On failure:

{
    "status": {
        "code": 500,
        "text": "...error text..."
    }
}

where 500 is HTTPStatus.INTERNAL_SERVER_ERROR.value.

class bh_database.base_table.WriteCapableTable(**kwargs)

Bases: ReadOnlyTable

Implement an abstract base model (table) class which INSERT, UPDATE and DELETE functionalities.

This table class should be the parent class for the majority of applications’ tables, whom the contents would change.

In general, for this abstract class and its abstract descendant classes, do not access query when not assigned. E.g.:

assert WriteCapableTable.query == None

It results in the following exception:

ArgumentError("Column expression, FROM clause, or other columns clause element expected, <class 'bh_database.core.BaseSQLAlchemy'>.")

run_execute_sql(sql: str, auto_session=False) → ResultStatus

Run an execute SQL full text statement and return a ResultStatus.

An execute SQL is an UPDATE or a DELETE SQL statement. It is assumed an execute SQL statement, there is no check enforced.

Parameters:

sql (str) – the full text execute SQL statement.
auto_session (bool) – upon an operation on the underlying database, SQLAlchemy auto starts a transaction if there is not one in progress. If this method is in a single call, the caller should set this param to True to get rid of the transaction when finished: if the call was successful, the transaction is committed, otherwise it rolls back. If it is called as a part of an ongoing transaction, in which case SQLAlchemy does not start another transaction, then just ignore this param, the caller is responsible for managing transaction atomicity.

Returns:

ResultStatus.

Further illustrations of return value, as a dictionary.

On successful:

{
    "status": {
        "code": 200,
        "text": ""
    }
}        

On failure:

{
    "status": {
        "code": 500,
        "text": "...message text..."
    }
}        

run_stored_proc(stored_proc_name: str, params: list, auto_session=False) → ResultStatus

Execute a stored procedure which returns some data.

It is assumed the stored procedure returns some data.

Parameters:

sql (stored_proc_name) – the name of the stored procedure.
params (list) – list of param values passed to the stored procedure.
auto_session (bool) – upon an operation on the underlying database, SQLAlchemy auto starts a transaction if there is not one in progress. If this method is in a single call, the caller should set this param to True to get rid of the transaction when finished: if the call was successful, the transaction is committed, otherwise it rolls back. If it is called as a part of an ongoing transaction, in which case SQLAlchemy does not start another transaction, then just ignore this param, the caller is responsible for managing transaction atomicity.

Returns:

ResultStatus.

Further illustrations of return value, as a dictionary.

On successful:

{
    "status": {
        "code": 200,
        "text": "Data has been retrieved successfully."
    },
    "data": [
        {...},
        ...
        {...}
    ]
}        

On failure:

{
    "status": {
        "code": 500,
        "text": "...error text..."
    }
}        

write_to_database(data: list) → ResultStatus

Write new records and modified records to the underlying database table.

When all data have been written, it will flush the transaction to cause any potential database violation to come out as an exception so that the result can be accurately determined, freeing the callers from having to handle any possible exception. Callers only have to check the returned result.

Flushing the transaction also causes intermediate pending committed data be available, callers can access these without having to call flush_transcation().

Assumptions:

The table has a single primary key of type integer.
The database already has table unique_id and stored method get_unique_id defined. Please see Database requirements for more detail.

Transaction:

callers must either call commit_transaction() or rollback_transaction() to commit or rollback the write respectively. E.g.:

Employees.begin_transaction(Employees)
status = Employees().write_to_database([new_emp1, new_emp2])
Employees.commit_transaction(Employees)

Parameters:

data (list) – data contains both new records and updated records.

An example of data:

 [
     {
         "col_1": 999, ..., "col_n": "xxx",
         "recStatus": "<new> | <modified>"
     },
     ...,
     {
         "col_1": 999, ..., "col_n": "xxx",
         "recStatus": "<new> | <modified>"
     },
]           

For each new record (row) in data, if primary key is present, and has a valid integer value, then a new unique integer Id is not requested. Otherwise, calls stored method get_unique_id with the table name and primary key column name to get next unique integer Id.

Returns:

ResultStatus.

Further illustrations of return value, as a dictionary.

On successful:

{
    "status": {
        "code": 200,
        "text": "Data has been saved successfully."
    },
    "{__tablename__}_new_list": [
        {...}, ... ,{}
    ],
    "{__tablename__}_updated_list": [
        {...}, ... ,{}
    ]
}

__tablename__: i.e. invoice, service, etc. E.g.:

service_new_list

service_updated_list

Either {__tablename__}_new_list or {__tablename__}_updated_list can be empty, but not both. At least one list must have a single object in it.

Record/row objects in these lists have recStatus removed.

On failure:

{
    "status": {
        "code": 500,
        "text": "...error text..."
    }
}

See the following test modules for more info:

./tests/test_25_base_table_crud_methods_postgresql.py

./tests/test_26_base_table_crud_methods_mysql.py