Core module

Core database wrapper classes for SQLAlchemy.

These classes currently support only two database types: MySQL and PostgreSQL. Drivers required, respectively:

https://pypi.org/project/mysql-connector-python/.

https://pypi.org/project/psycopg2/.

Later classes, see module Base Table classes, which implement base models (tables) that provide wrapper methods to run full SQL statements, and stored methods. There are some differences between drivers on stored methods execution, that is why only MySQL and PostgreSQL are supported presently. Internally, when executing stored methods, the codes use the connection URL to determine which database it is to extract result set.

Classes in this module provide the following functionalities:

Database connection management. See Database.

A custom SQLAlchemy query class which implements paginating. See BaseQuery and Paginator.

A generic BaseSQLAlchemy base model (table), which should be the indirect base model for applications’ models. This class encapsulates:

SQLAlchemy scoped session, thereby providing methods to implement transaction atomicity.

A custom base query with paginating ability mentioned above.

Fully implemented dunder methods str() and repr(), which are implemented by class BaseModel: one of the base classes of BaseSQLAlchemy.

A generic SQLAlchemy declarative base class Base which should include all required metaclasses.

Applications should not be directly interested in the BaseSQLAlchemy base model (table). Module Base Table classes defines two application-ready base model classes base_table.ReadOnlyTable and base_table.WriteCapableTable. Applications’ models should descend from either one of these as appropriate.

From this module, the only class which applications should call directly is Database. To connect to a database server, applications should not have do not more than:

# Getting rid of any residual connection. If we are absolutely sure there is none, 
# then there is no need for this call.
Database.disconnect()
Database.connect(db_url, [schema | None])

class bh_database.core.Base(**kwargs: Any)

Bases: DeclarativeBase

metaclass: alias of DeclarativeMeta

metadata: ClassVar[MetaData] = MetaData(): Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

orm_declarative_metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>: Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class bh_database.core.BaseQuery(entities: _ColumnsClauseArgument[Any] | Sequence[_ColumnsClauseArgument[Any]], session: Session | None = None)

Bases: Query

Custom base query class.

Provide pagination capability for SQLAlchemy Query Object.

All models (tables) descend from BaseSQLAlchemy will have their class attribute query set to this custom query class, therefore, automatically has pagination capability.

paginate(page: int, per_page: int) → Paginator

Pagination method.

Parameters:

page (int) – the page number to retrieve data for.
per_page (int) – how many records to retrieve for each page.

Returns:

a paginator.Paginator instance.

class bh_database.core.BaseModel

Bases: object

A custom base model / table class for SQLAlchemy declarative base model.

All models (tables) descend from BaseSQLAlchemy will have this custom base model as one of the base classes.

This class basically implements two (2) magic methods: object.__str__(self); and object.__repr__(self).

For an usage example, see ./tests/test_20_base_table_dunder.py.

class bh_database.core.BaseSQLAlchemy(**kwargs: Any)

Bases: Base, BaseModel

The parent most class for all models / tables.

Provide methods to implement transaction atomicity.

Class attributes:: session = None. When set, is of type sqlalchemy.orm.session.Session. This attribute is set after successfully calling the Database’s connect() method.

query = None. When set, is of type BaseQuery. This attribute is set after successfully calling the Database’s connect() method.

Application models, i.e. tables, descend indirectly from this class, and hence inherits attributes session and query. Table classes use these two class attributes to talk to the connected database: they should not need the Database class for anything else.

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

assert core.BaseSQLAlchemy.query == None

It results in the following exception:

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

session = None: Class attribute. When set, is of type sqlalchemy.orm.session.Session. This attribute is set after successfully calling the Database’s connect() method.

query = None: Class attribute. When set, is of type BaseQuery. This attribute is set after successfully calling the Database’s connect() method.

begin_transaction(): Start a new transaction.

flush_transaction()

Flush an ongoing transaction.

Make intermediate results available so that they can be accessed while the transaction is still on going. An example of intermediate result is a new unique key value, this value might be needed as a foreign key for detail records awaiting written.

Note on Exception:

Potential unhandled exception: caller must handle the exception.

Within an ongoing transaction, if there is any potential database violation, calling this method will cause the driver to raise exception on the violation.

commit_transaction()

Commit a current transaction.

Note on Exception:

Potential unhandled exception: caller must handle the exception.

Within an ongoing transaction, if there is any potential database violation, calling this method will cause the driver to raise exception on the violation.

It is recommended to call finalise_transaction() instead.

rollback_transaction()

Rollback a current transaction.

When absolutely certain that the ongoing transaction must be rolled back, then call this method. Otherwise it is recommended to call finalise_transaction() instead.

finalise_transaction(status: ResultStatus)

Commit or rollback a transaction based on status.code.

Parameters:: status (ResultStatus) – result of the last CRUD call. ResultStatus.

class bh_database.core.DatabaseType(value)

Bases: Enum

Enumerated constants identifying supported databases.

Define some enumerations for the supported database servers.

Unknown = -1

MySQL = 1

PostgreSQL = 2

class bh_database.core.Database

Bases: object

Provide database connection management.

Applications should not instantiate this class. It is semantically a static class.

All applications proper models (tables) should descend from BaseSQLAlchemy. The connect() method sets both BaseSQLAlchemy class attributes session and query, armed with these attributes, application models have sufficient means to carry out CRUD operations.

Thus, to connect to a database server, applications should not have do not more than:

# Getting rid of any residual connection. If we are absolutely sure there is none, 
# then there is no need for this call.
Database.disconnect()
Database.connect(db_url, [schema | None])

See the following tests for more info:

./tests/test_05_core_basesqlalchemy_postgresql.py

./tests/test_06_core_basesqlalchemy_mysql.py

Class attributes:: engine = None. When set, is of type sqlalchemy.future.engine.Engine.

session_factory = None. When set, is of type sqlalchemy.orm.sessionmaker.

database_session = None. When set, is of type sqlalchemy.orm.scoping.scoped_session.

For a usage example, see ./tests/test_01_core_database_postgresql.py and ./tests/test_02_core_database_mysql.py.

engine = None: Class attribute. When set, is of type sqlalchemy.future.engine.Engine.

session_factory = None: Class attribute. When set, is of type sqlalchemy.orm.sessionmaker.

database_session = None: Class attribute. When set, is of type sqlalchemy.orm.scoping.scoped_session.

static database_type(db_url=None) → DatabaseType

Return an enum which represents a supported database server type.

Parameters:: db_url (str) – optional. A valid database connection string. If not specified, then the current database connection string is used. That means, if this param is specified, then a valid database connection is not required.
Returns:: a DatabaseType’s enum.
Raises:: AttributeError – if not connected to a database, i.e. invalid database connection, and param db_url is not specified.

static driver_name() → str

Return the driver name of the current database connection.

Returns:: The driver name of the current database connection. E.g. if the connection string is "postgresql+psycopg2://postgres:<password>@localhost/employees", then the returned value is postgresql+psycopg2.
Return type:: str.
Raises:: AttributeError – if not connected to a database, i.e. invalid database connection.

static connect(db_url: str, schema: str) → None

Establish a connection to a database server.

Parameters:

db_url (str) – a valid database connection string.
schema (str) – the database schema in the database to connect to. Presently only required if connecting to a PostgreSQL database.

Create the following class attributes engine, session_factory and scoped session database_session.

Also, set both BaseSQLAlchemy’s session and query as:

BaseSQLAlchemy.session = Database.database_session(future=True)
BaseSQLAlchemy.query = Database.database_session.query_property(BaseQuery)

BaseSQLAlchemy’s query is still None after this assignment. IT IS POSTULATING THAT, this is because BaseSQLAlchemy is abstract, which means it does not have an associated database table declared. Postulating because there is not an official document confirms that this is the case. See the following test modules for more info:

./tests/test_15_base_table_postgresql.py

./tests/test_16_base_table_mysql.py

Application models, i.e. tables, descend indirectly from BaseSQLAlchemy, and hence inherits class attributes session and query. Table classes use these two class attributes to talk to the connected database: they should not need the Database class for anything else.

static disconnect() → None

Disconnect from database.

Remove the scoped session, close all connections, dispose the connection pool used by the engine (i.e. calling sqlalchemy.engine.Engine.dispose(close: bool = True) -> None).

Then set class attributes database_session, session_factory and engine to None.

Finally, set both BaseSQLAlchemy’s session and query class attributes to None also.

Note, any scoped sessions, queries, etc. created locally via database_session are still valid after calling this method. That is:

POSTGRESQL_DB_URL = "postgresql+psycopg2://postgres:<password>@localhost/employees"
POSTGRESQL_DB_SCHEMA = "employees"

SELECT_EMPLOYEES = ("select * from employees where (upper(last_name) like '%NAS%')" 
    " and (upper(first_name) like '%AN') order by emp_no;")

Database.connect(POSTGRESQL_DB_URL, POSTGRESQL_DB_SCHEMA)

session = Database.database_session()
query = Database.database_session.query_property(BaseQuery)
other_session = Database.database_session()

Database.disconnect()

assert session != None
assert query != None
assert other_session != None

result = session.execute(text(SELECT_EMPLOYEES))
# result is CursorResult.
assert result.rowcount == 38

result = other_session.execute(text(SELECT_EMPLOYEES))
# result is CursorResult.
assert result.rowcount == 38

Core test modules

There are following modules:

./tests/test_01_core_database_postgresql.py
./tests/test_02_core_database_mysql.py
./tests/test_05_core_basesqlalchemy_postgresql.py
./tests/test_06_core_basesqlalchemy_mysql.py
./tests/test_11_paginator_postgresql.py
./tests/test_12_paginator_mysql.py
./tests/test_15_base_table_postgresql.py
./tests/test_16_base_table_mysql.py
./tests/test_17_base_table_methods.py
./tests/test_20_base_table_dunder.py

These test modules, together with fixtures defined in ./tests/conftest.py should illustrate how the classes in this core module should work together.