trulens.core.database.utils¶

trulens.core.database.utils ¶

is_legacy_sqlite(engine: Engine) -> bool

Check if DB is an existing file-based SQLite created with the legacy LocalSQLite implementation.

This database was removed since trulens 0.29.0 .

is_memory_sqlite(
    engine: Optional[Engine] = None,
    url: Optional[Union[URL, str]] = None,
) -> bool

Check if DB is an in-memory SQLite instance.

Either engine or url can be provided.

check_db_revision(
    engine: Engine,
    prefix: str = core_db.DEFAULT_DATABASE_PREFIX,
    prior_prefix: Optional[str] = None,
)

Check if database schema is at the expected revision.

PARAMETER	DESCRIPTION
`engine`	SQLAlchemy engine to check. TYPE: `Engine`
`prefix`	Prefix used for table names including alembic_version in the current code. TYPE: `str` DEFAULT: `DEFAULT_DATABASE_PREFIX`
`prior_prefix`	Table prefix used in the previous version of the database. Before this configuration was an option, the prefix was equivalent to "". TYPE: `Optional[str]` DEFAULT: `None`

coerce_ts(ts: Union[datetime, str, int, float]) -> datetime

Coerce various forms of timestamp into datetime.

copy_database(
    src_url: str,
    tgt_url: str,
    src_prefix: str,
    tgt_prefix: str,
)

Copy all data from a source database to an EMPTY target database.

Important considerations:

All source data will be appended to the target tables, so it is important that the target database is empty.
Will fail if the databases are not at the latest schema revision. That can be fixed with TruSession(database_url="...", database_prefix="...").migrate_database()
Might fail if the target database enforces relationship constraints, because then the order of inserting data matters.
This process is NOT transactional, so it is highly recommended that the databases are NOT used by anyone while this process runs.