hdx.database
Database utilities
Database Objects
class Database()
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 namehost
Optional[str] - Host where database is locatedport
Union[int, str, None] - Database portusername
Optional[str] - Username to log into databasepassword
Optional[str] - Password to log into databasedialect
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 belowrecreate_schema
bool - Whether to recreate schemaschema_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 usernamessh_password
str - SSH passwordssh_private_key
- Location of SSH private key (instead of ssh_password) For more advanced usage, see SSHTunnelForwarder documentation.
get_engine
def get_engine() -> Engine
Returns SQLAlchemy engine.
Returns:
sqlalchemy.Engine
- SQLAlchemy engine
get_session
def get_session() -> Session
Returns SQLAlchemy session.
Returns:
sqlalchemy.orm.Session
- SQLAlchemy session
get_reflected_classes
def get_reflected_classes() -> Any
Gets reflected classes.
Returns:
Any
- Reflected classes
get_prepare_results
def get_prepare_results() -> Any
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]]
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
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
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]
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