trulens.core.database.sqlalchemy¶

trulens.core.database.sqlalchemy ¶

Classes¶

SQLAlchemyDB ¶

Bases: DB

Database implemented using sqlalchemy.

See abstract class DB for method reference.

Attributes¶

redact_keys `class-attribute` `instance-attribute` ¶

redact_keys: bool = DEFAULT_DATABASE_REDACT_KEYS

Redact secrets before writing out data.

table_prefix `class-attribute` `instance-attribute` ¶

table_prefix: str = DEFAULT_DATABASE_PREFIX

The prefix to use for all table names.

DB interface requirement.

engine_params `class-attribute` `instance-attribute` ¶

engine_params: dict = Field(default_factory=dict)

SQLAlchemy-related engine params.

session_params `class-attribute` `instance-attribute` ¶

session_params: dict = Field(default_factory=dict)

SQLAlchemy-related session.

engine `class-attribute` `instance-attribute` ¶

engine: Optional[Engine] = None

SQLAlchemy engine.

session `class-attribute` `instance-attribute` ¶

session: Optional[sessionmaker] = None

SQLAlchemy session(maker).

orm `instance-attribute` ¶

orm: Type[ORM]

Container of all the ORM classes for this database.

This should be set to a subclass of ORM upon initialization.

Functions¶

__rich_repr__ ¶

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

str ¶

__str__() -> str

Relatively concise identifier string for this instance.

from_tru_args `classmethod` ¶

from_tru_args(
    database_url: Optional[str] = None,
    database_engine: Optional[Engine] = None,
    database_redact_keys: Optional[
        bool
    ] = DEFAULT_DATABASE_REDACT_KEYS,
    database_prefix: Optional[
        str
    ] = DEFAULT_DATABASE_PREFIX,
    **kwargs: Dict[str, Any]
) -> SQLAlchemyDB

Process database-related configuration provided to the Tru class to create a database.

Emits warnings if appropriate.

from_db_url `classmethod` ¶

from_db_url(
    url: str, **kwargs: Dict[str, Any]
) -> SQLAlchemyDB

Create a database for the given url.

PARAMETER	DESCRIPTION
`url`	The database url. This includes database type. TYPE: `str`
`kwargs`	Additional arguments to pass to the database constructor. TYPE: `Dict[str, Any]` DEFAULT: `{}`

RETURNS	DESCRIPTION
`SQLAlchemyDB`	A database instance.

from_db_engine `classmethod` ¶

from_db_engine(
    engine: Engine, **kwargs: Dict[str, Any]
) -> SQLAlchemyDB

Create a database for the given engine. Args: engine: The database engine. kwargs: Additional arguments to pass to the database constructor. Returns: A database instance.

check_db_revision ¶

check_db_revision()

See DB.check_db_revision.

migrate_database ¶

migrate_database(prior_prefix: Optional[str] = None)

See DB.migrate_database.

reset_database ¶

reset_database()

See DB.reset_database.

insert_record ¶

insert_record(record: Record) -> RecordID

See DB.insert_record.

batch_insert_record ¶

batch_insert_record(
    records: List[Record],
) -> List[RecordID]

See DB.batch_insert_record.

get_app ¶

get_app(app_id: AppID) -> Optional[JSONized]

See DB.get_app.

update_app_metadata ¶

update_app_metadata(
    app_id: AppID, metadata: Dict[str, Any]
) -> Optional[AppDefinition]

See DB.update_app_metadata.

get_apps ¶

get_apps(
    app_name: Optional[AppName] = None,
) -> Iterable[JSON]

See DB.get_apps.

insert_app ¶

insert_app(app: AppDefinition) -> AppID

See DB.insert_app.

delete_app ¶

delete_app(app_id: AppID) -> None

Deletes an app from the database based on its app_id.

PARAMETER	DESCRIPTION
`app_id`	The unique identifier of the app to be deleted. TYPE: `AppID`

insert_feedback_definition ¶

insert_feedback_definition(
    feedback_definition: FeedbackDefinition,
) -> FeedbackDefinitionID

See DB.insert_feedback_definition.

get_feedback_defs ¶

get_feedback_defs(
    feedback_definition_id: Optional[
        FeedbackDefinitionID
    ] = None,
) -> DataFrame

See DB.get_feedback_defs.

insert_feedback ¶

insert_feedback(
    feedback_result: FeedbackResult,
) -> FeedbackResultID

See DB.insert_feedback.

batch_insert_feedback ¶

batch_insert_feedback(
    feedback_results: List[FeedbackResult],
) -> List[FeedbackResultID]

See DB.batch_insert_feedback.

get_feedback_count_by_status ¶

get_feedback_count_by_status(
    record_id: Optional[RecordID] = None,
    feedback_result_id: Optional[FeedbackResultID] = None,
    feedback_definition_id: Optional[
        FeedbackDefinitionID
    ] = None,
    status: Optional[
        Union[
            FeedbackResultStatus,
            Sequence[FeedbackResultStatus],
        ]
    ] = None,
    last_ts_before: Optional[datetime] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
    shuffle: bool = False,
    run_location: Optional[FeedbackRunLocation] = None,
) -> Dict[FeedbackResultStatus, int]

See DB.get_feedback_count_by_status.

get_feedback ¶

get_feedback(
    record_id: Optional[RecordID] = None,
    feedback_result_id: Optional[FeedbackResultID] = None,
    feedback_definition_id: Optional[
        FeedbackDefinitionID
    ] = None,
    status: Optional[
        Union[
            FeedbackResultStatus,
            Sequence[FeedbackResultStatus],
        ]
    ] = None,
    last_ts_before: Optional[datetime] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
    shuffle: Optional[bool] = False,
    run_location: Optional[FeedbackRunLocation] = None,
) -> DataFrame

See DB.get_feedback.

get_records_and_feedback ¶

get_records_and_feedback(
    app_ids: Optional[List[str]] = None,
    app_name: Optional[AppName] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
) -> Tuple[DataFrame, Sequence[str]]

See DB.get_records_and_feedback.

insert_ground_truth ¶

insert_ground_truth(
    ground_truth: GroundTruth,
) -> GroundTruthID

See DB.insert_ground_truth.

batch_insert_ground_truth ¶

batch_insert_ground_truth(
    ground_truths: List[GroundTruth],
) -> List[GroundTruthID]

See DB.batch_insert_ground_truth.

get_ground_truth ¶

get_ground_truth(
    ground_truth_id: str | None = None,
) -> Optional[JSONized]

See DB.get_ground_truth.

get_ground_truths_by_dataset ¶

get_ground_truths_by_dataset(
    dataset_name: str,
) -> DataFrame | None

See DB.get_ground_truths_by_dataset.

insert_dataset ¶

insert_dataset(dataset: Dataset) -> DatasetID

See DB.insert_dataset.

get_datasets ¶

get_datasets() -> DataFrame

See DB.get_datasets.

AppsExtractor ¶

Utilities for creating dataframes from orm instances.

Functions¶

get_df_and_cols ¶

get_df_and_cols(
    apps: Optional[List["db_orm.ORM.AppDefinition"]] = None,
    records: Optional[List["db_orm.ORM.Record"]] = None,
) -> Tuple[DataFrame, Sequence[str]]

Produces a records dataframe which joins in information from apps and feedback results.

PARAMETER	DESCRIPTION
`apps`	If given, includes all records of all of the apps in this iterable. TYPE: `Optional[List['db_orm.ORM.AppDefinition']]` DEFAULT: `None`
`records`	If given, includes only these records. Mutually exclusive with `apps`. TYPE: `Optional[List['db_orm.ORM.Record']]` DEFAULT: `None`

extract_apps ¶

extract_apps(
    apps: Iterable["db_orm.ORM.AppDefinition"],
    records: Optional[List["db_orm.ORM.Record"]] = None,
) -> Iterable[DataFrame]

Creates record rows with app information.

TODO: The means for enumerating records in this method is not ideal as it does a lot of filtering.

trulens.core.database.sqlalchemy¶