Skip to content

hdx.database

Database utilities

Database Objects

class Database()

[view_source]

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.

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.

Arguments:

  • 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 - See below
  • 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.

get_engine

def get_engine() -> Engine

[view_source]

Returns SQLAlchemy engine.

Returns:

  • sqlalchemy.Engine - SQLAlchemy engine

get_session

def get_session() -> Session

[view_source]

Returns SQLAlchemy session.

Returns:

  • sqlalchemy.orm.Session - SQLAlchemy session

get_reflected_classes

def get_reflected_classes() -> Any

[view_source]

Gets reflected classes.

Returns:

  • Any - Reflected classes

get_prepare_results

def get_prepare_results() -> Any

[view_source]

Returns results from prepare function.

Returns:

  • Any - Results from prepare function

create_session

@staticmethod
def create_session(
        engine: Optional[Engine] = None,
        db_uri: Optional[str] = None,
        table_base: Type[DeclarativeBase] = NoTZBase,
        reflect: bool = False) -> Tuple[Session, Optional[Properties]]

[view_source]

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.

Arguments:

  • 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, Optional[Properties]]: (SQLAlchemy session, reflected classes if available)

recreate_schema

@staticmethod
def recreate_schema(engine: Engine, schema_name: str = "public") -> bool

[view_source]

Wipe and create empty schema in database using SQLAlchemy.

Arguments:

  • engine Engine - SQLAlchemy engine to use.
  • schema_name str - Schema name. Defaults to "public".

Returns:

  • bool - True if all successful, False if not

prepare_view

@staticmethod
def prepare_view(view_params: Dict) -> TableClause

[view_source]

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

Arguments:

  • view_params Dict - Dictionary with keys name, metadata, selectable

Returns:

  • TableClause - SQLAlchemy View

prepare_views

@classmethod
def prepare_views(cls, view_params_list: List[Dict]) -> List[TableClause]

[view_source]

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

Arguments:

  • view_params_list List[Dict] - List of dictionaries with view parameters

Returns:

  • List[TableClause] - SQLAlchemy Views