DatabaseContext¶

class felis.db.utils.DatabaseContext(metadata, engine)¶

Manage the database connection and SQLAlchemy metadata.

Parameters:

metadata (MetaData) – The SQLAlchemy metadata object.
engine (Engine | MockConnection) – The SQLAlchemy engine or mock connection object.

Methods Summary

`create_all`()	Create all tables in the schema using the metadata object.
`create_mock_engine`(engine_url[, output_file])	Create a mock engine for testing or dumping DDL statements.
`drop`()	Drop the schema in the database if it exists.
`execute`(statement)	Execute a SQL statement on the engine and return the result.
`initialize`()	Create the schema in the database if it does not exist.

Methods Documentation

create_all()¶

Create all tables in the schema using the metadata object.

static create_mock_engine(engine_url, output_file=None)¶

Create a mock engine for testing or dumping DDL statements.

Parameters:

engine_url (str | URL) – The SQLAlchemy engine URL.
output_file (Optional[IO[str]], default: None) – The file to write the SQL statements to. If None, the statements will be written to stdout.

Returns:

The mock connection object.

Return type:

sqlalchemy.engine.mock.MockConnection

drop()¶

Drop the schema in the database if it exists.

Notes

In MySQL, this will drop a database. In PostgreSQL, it will drop a schema. For other variants, this is an unsupported operation.

execute(statement)¶

Execute a SQL statement on the engine and return the result.

Notes

This is just a wrapper around the execution method of the connection object, which may execute on a real or mock connection.

initialize()¶

Create the schema in the database if it does not exist.

Raises:

ValueError – Raised if the database is not supported or it already exists.
sqlalchemy.exc.SQLAlchemyError – Raised if there is an error creating the schema.

Return type:

None

Notes

In MySQL, this will create a new database and, in PostgreSQL, it will create a new schema. For other variants, this is an unsupported operation.