Core Module
Core database wrapper classes for SQLAlchemy.
These classes currently support only two database types: MySQL and PostgreSQL. Drivers required, respectively:
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
andPaginator
.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 ofBaseSQLAlchemy
.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 attributequery
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)
-
The parent most class for all models / tables.
Provide methods to implement transaction atomicity.
- Class attributes:
Application models, i.e. tables, descend indirectly from this class, and hence inherits attributes
session
andquery
. Table classes use these two class attributes to talk to the connected database: they should not need theDatabase
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
’sconnect()
method.
- query = None
Class attribute. When set, is of type
BaseQuery
. This attribute is set after successfully calling theDatabase
’sconnect()
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
. Theconnect()
method sets bothBaseSQLAlchemy
class attributessession
andquery
, 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 ispostgresql+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 sessiondatabase_session
.Also, set both
BaseSQLAlchemy
’ssession
andquery
as:BaseSQLAlchemy.session = Database.database_session(future=True) BaseSQLAlchemy.query = Database.database_session.query_property(BaseQuery)
BaseSQLAlchemy
’squery
is stillNone
after this assignment. IT IS POSTULATING THAT, this is becauseBaseSQLAlchemy
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 attributessession
andquery
. Table classes use these two class attributes to talk to the connected database: they should not need theDatabase
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
andengine
toNone
.Finally, set both
BaseSQLAlchemy
’ssession
andquery
class attributes toNone
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:
These test modules, together with fixtures defined in ./tests/conftest.py
should illustrate how the classes in this core module should work together.