ensembl.utils.database package

Database module.

Submodules

ensembl.utils.database.dbconnection module

Database connection handler.

This module provides the main class to connect to and access databases. It will be an ORM-less connection, that is, the data can only be accessed via SQL queries (see example below).

Examples

>>> from ensembl.utils.database import DBConnection
>>> dbc = DBConnection("mysql://ensro@mysql-server:4242/mydb")
>>> # You can access the database data via sql queries, for instance:
>>> results = dbc.execute("SELECT * FROM my_table;")
>>> # Or via a connection in a transaction manner:
>>> with dbc.begin() as conn:
>>>     results = conn.execute("SELECT * FROM my_table;")

class ensembl.utils.database.dbconnection.DBConnection(url, reflect=True, **kwargs)

Bases: object

Database connection handler, providing also the database’s schema and properties.

Parameters:

• url (TypeVar(StrURL, str, URL)) – URL to the database, e.g. mysql://user:passwd@host:port/my_db.

• reflect (bool, default: True) – Reflect the database schema or not.

__repr__()

Returns a string representation of this object.

Return type:

str

begin(*args)

Returns a context manager delivering a database connection with a transaction established.

Return type:

ContextManager[Connection]

connect()

Returns a new database connection.

Return type:

Connection

create_all_tables(metadata)

Create the tables from the metadata and set the metadata.

This assumes the database is empty beforehand. If the tables already exist, they will be ignored. If there are other tables, you may need to run self.load_metadata() to update the metadata schema.

Return type:

None

create_table(table)

Create a table in the database and update the metadata. Do nothing if the table already exists.

Return type:

None

property db_name: str | None

Returns the database name.

property dialect: str

Returns the SQLAlchemy database dialect name of the database host.

dispose()

Disposes of the connection pool.

Return type:

None

get_columns(table)

Returns the column names for the given table.

Parameters:

table (str) – Table name.

Return type:

list[str]

get_primary_key_columns(table)

Returns the primary key column names for the given table.

Parameters:

table (str) – Table name.

Return type:

list[str]

property host: str | None

Returns the database host.

load_metadata()

Loads the metadata information of the database.

Return type:

None

property port: int | None

Returns the port of the database host.

session_scope()

Provides a transactional scope around a series of operations with rollback in case of failure.

Bear in mind MySQL’s storage engine MyISAM does not support rollback transactions, so all the modifications performed to the database will persist.

Return type:

Generator[Session, None, None]

property tables: dict[str, Table]

Returns the database tables keyed to their name, or an empty dict if no metadata was loaded.

test_session_scope()

Provides a transactional scope around a series of operations that will be rolled back at the end.

Bear in mind MySQL’s storage engine MyISAM does not support rollback transactions, so all the modifications performed to the database will persist.

Return type:

Generator[Session, None, None]

property url: str

Returns the database URL.

ensembl.utils.database.unittestdb module

Unit testing database handler.

This module provides the main class to create and drop testing databases, populating them from preexisting dumps (if supplied).

Examples

>>> from ensembl.utils.database import UnitTestDB
>>> # For more safety use the context manager (automatically drops the database even if things go wrong):
>>> with UnitTestDB("mysql://user:passwd@mysql-server:4242/", "path/to/dumps", "my_db") as test_db:
>>>    dbc = test_db.dbc

>>> # If you know what you are doing you can also control when the test_db is dropped:
>>> test_db = UnitTestDB("mysql://user:passwd@mysql-server:4242/", "path/to/dumps", "my_db")
>>> # You can access the database via test_db.dbc, for instance:
>>> dbc = test_db.dbc
>>> # At the end do not forget to drop the database
>>> test_db.drop()

class ensembl.utils.database.unittestdb.UnitTestDB(server_url, *, dump_dir=None, name=None, metadata=None, tmp_path=None)

Bases: object

Creates and connects to a new test database, applying the schema and importing the data.

Parameters:

• server_url (TypeVar(StrURL, str, URL)) – URL of the server hosting the database.

• metadata (Optional[MetaData], default: None) – Use this metadata to create the schema instead of using an SQL schema file.

• dump_dir (Union[str, PathLike[str], None], default: None) – Directory path with the database schema in table.sql (mandatory) and one TSV data file (without headers) per table following the convention <table_name>.txt (optional).

• name (Optional[str], default: None) – Name to give to the new database. If not provided, the last directory name of dump_dir will be used instead. In either case, the new database name will be prefixed by the username.

• tmp_path (Union[str, PathLike[str], None], default: None) – Temp dir where the test db is created if using SQLite (otherwise use current dir).

Variables:

dbc – Database connection handler.

Raises:

FileNotFoundError – If table.sql is not found.

__repr__()

Returns a string representation of this object.

Return type:

str

drop()

Drops the database.

Return type:

None