hdx.database.Database

class Database(engine: Engine | None = None, db_uri: str | None = None, database: str | None = None, host: str | None = None, port: int | str | None = None, username: str | None = None, password: str | None = None, dialect: str = 'postgresql', driver: str | None = None, db_has_tz: bool = False, table_base: type[DeclarativeBase] | None = None, reflect: bool = False, **kwargs: Any)

Database helper class to handle ssh tunnels, waiting for PostgreSQL to be up etc. Can be used in a with statement returning a Database object that if reflect is True will have a variable reflected_classes containing the reflected classes. Either an SQLAlchemy engine, database uri or various database parameters must be supplied.

db_has_tz which defaults to False indicates whether database datetime columns have timezones. If not, conversion occurs between Python datetimes with timezones to timezoneless database columns (but not when using reflection). If table_base is supplied, db_has_tz is ignored.

The database can be restored from a given pg_restore file supplied in the parameter pg_restore_file.

There is an option to wipe and create an empty schema in the database by setting recreate_schema to True and setting a schema_name ("public" is the default).

If a prepare function is supplied in prepare_fn, it will be executed before Base.metadata.create_all and the results of it returned in instance variable prepare_results.

Parameters

• engine : Optional[Engine] — SQLAlchemy engine to use.

• db_uri : Optional[str] — Connection URI.

• database : Optional[str] — Database name

• host : Optional[str] — Host where database is located

• port : Union[int, str, None] — Database port

• username : Optional[str] — Username to log into database

• password : Optional[str] — Password to log into database

• dialect : str — Database dialect. Defaults to "postgresql".

• driver : Optional[str] — Database driver. Defaults to None (psycopg if postgresql or None)

• db_has_tz : bool — True if db datetime columns have timezone. Defaults to False.

• table_base : Optional[Type[DeclarativeBase]] — Override table base. Defaults to None.

• reflect : bool — Whether to reflect existing tables. Defaults to False.

• **kwargs : Any — See below

• pg_restore_file : Path | str — Restore database from pg_restore file

• recreate_schema : bool — Whether to recreate schema

• schema_name : str — Database schema name. Defaults to "public".

• prepare_fn : Callable[[], None]] — Function to call before Base.metadata.create_all.

• ssh_host : str — SSH host (the server to connect to)

• ssh_port : int — SSH port. Defaults to 22.

• ssh_username : str — SSH username

• ssh_password : str — SSH password

• ssh_private_key — Location of SSH private key (instead of ssh_password)

• For more advanced usage, see SSHTunnelForwarder documentation.

Methods

• cleanup — Cleanup SQLAlchemy.

• drop_all — Drop all tables.

• get_engine — Returns SQLAlchemy engine.

• get_session — Returns SQLAlchemy session.

• get_reflected_classes — Gets reflected classes.

• get_prepare_results — Returns results from prepare function.

• batch_populate — Batch populate database table.

• create_session — Creates SQLAlchemy session given SQLAlchemy engine or database uri (one of which must be supplied). Tables must inherit from Base in hdx.utilities.database unless base is defined. If reflect is True, classes will be reflected from an existing database and the reflected classes will be returned. Note that type annotation maps don't work with reflection.

• drop_schema — Wipe schema in database using SQLAlchemy.

• recreate_schema — Wipe and create empty schema in database using SQLAlchemy.

• prepare_view — Prepare SQLAlchemy view from dictionary with keys: name, metadata and selectable. Must be run before Base.metadata.create_all.

• prepare_views — Prepare SQLAlchemy views from a list of dictionaries with keys: name, metadata and selectable. Must be run before Base.metadata.create_all.

method Database.cleanup() → None

Cleanup SQLAlchemy.

Returns

• sqlalchemy.Engine — SQLAlchemy engine

method Database.drop_all() → None

Drop all tables.

Returns

• None — None

method Database.get_engine() → Engine

Returns SQLAlchemy engine.

Returns

• sqlalchemy.Engine — SQLAlchemy engine

method Database.get_session() → Session

Returns SQLAlchemy session.

Returns

• sqlalchemy.orm.Session — SQLAlchemy session

method Database.get_reflected_classes() → Any

Gets reflected classes.

Returns

• Any — Reflected classes

method Database.get_prepare_results() → Any

Returns results from prepare function.

Returns

• Any — Results from prepare function

method Database.batch_populate(rows: list[dict], dbtable: type[DeclarativeBase], batch_size: int = 1000) → None

Batch populate database table.

Parameters

• rows : List[Dict] — List of rows

• dbtable : Type[DeclarativeBase] — Database table

• batch_size : int — Batch size. Defaults to 1000.

Returns

• Tuple[Session, Any] — (SQLAlchemy session, base)

staticmethod Database.create_session(engine: Engine | None = None, db_uri: str | None = None, table_base: type[DeclarativeBase] = NoTZBase, reflect: bool = False) → tuple[Session, Any]

Creates SQLAlchemy session given SQLAlchemy engine or database uri (one of which must be supplied). Tables must inherit from Base in hdx.utilities.database unless base is defined. If reflect is True, classes will be reflected from an existing database and the reflected classes will be returned. Note that type annotation maps don't work with reflection.

Parameters

• engine : Optional[Engine] — SQLAlchemy engine to use. Defaults to None (create from db_uri).

• db_uri : Optional[str] — Connection URI. Defaults to None (use engine).

• table_base : Type[DeclarativeBase] — Base database table class. Defaults to NoTZBase.

• reflect : bool — Whether to reflect existing tables. Defaults to False.

Returns

• Tuple[Session, Any] — (SQLAlchemy session, base)

Raises

• DatabaseError

staticmethod Database.drop_schema(engine: Engine, schema_name: str = 'public') → bool

Wipe schema in database using SQLAlchemy.

Parameters

• engine : Engine — SQLAlchemy engine to use.

• schema_name : str — Schema name. Defaults to "public".

Returns

• bool — True if all successful, False if not

staticmethod Database.recreate_schema(engine: Engine, schema_name: str = 'public') → bool

Wipe and create empty schema in database using SQLAlchemy.

Parameters

• engine : Engine — SQLAlchemy engine to use.

• schema_name : str — Schema name. Defaults to "public".

Returns

• bool — True if all successful, False if not

staticmethod Database.prepare_view(view_params: dict) → TableClause

Prepare SQLAlchemy view from dictionary with keys: name, metadata and selectable. Must be run before Base.metadata.create_all.

Parameters

• view_params : Dict — Dictionary with keys name, metadata, selectable

Returns

• TableClause — SQLAlchemy View

classmethod Database.prepare_views(view_params_list: list[dict]) → list[TableClause]

Prepare SQLAlchemy views from a list of dictionaries with keys: name, metadata and selectable. Must be run before Base.metadata.create_all.

Parameters

• view_params_list : List[Dict] — List of dictionaries with view parameters

Returns

• List[TableClause] — SQLAlchemy Views