trulens.core.database.base¶

trulens.core.database.base ¶

Attributes¶

DEFAULT_DATABASE_PREFIX `module-attribute` ¶

DEFAULT_DATABASE_PREFIX: str = 'trulens_'

Default prefix for table names for trulens to use.

This includes alembic's version table.

DEFAULT_DATABASE_FILE `module-attribute` ¶

DEFAULT_DATABASE_FILE: str = 'default.sqlite'

Filename for default sqlite database.

The sqlalchemy url for this default local sqlite database is sqlite:///default.sqlite.

DEFAULT_DATABASE_REDACT_KEYS `module-attribute` ¶

DEFAULT_DATABASE_REDACT_KEYS: bool = False

Default value for option to redact secrets before writing out data to database.

Classes¶

DB ¶

Bases: SerialModel, ABC, WithIdentString

Abstract definition of databases used by trulens.

SQLAlchemyDB is the main and default implementation of this interface.

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

Prefix for table names for trulens to use.

May be useful in some databases where trulens is not the only app.

Functions¶

__rich_repr__ ¶

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

reset_database `abstractmethod` ¶

reset_database()

Delete all data.

migrate_database `abstractmethod` ¶

migrate_database(prior_prefix: Optional[str] = None)

Migrate the stored data to the current configuration of the database.

PARAMETER	DESCRIPTION
`prior_prefix`	If given, the database is assumed to have been reconfigured from a database with the given prefix. If not given, it may be guessed if there is only one table in the database with the suffix `alembic_version`. TYPE: `Optional[str]` DEFAULT: `None`

check_db_revision `abstractmethod` ¶

check_db_revision()

Check that the database is up to date with the current trulens version.

RAISES	DESCRIPTION
`ValueError`	If the database is not up to date.

get_db_revision `abstractmethod` ¶

get_db_revision() -> Optional[str]

Get the current revision of the database.

RETURNS	DESCRIPTION
`Optional[str]`	The current revision of the database.

insert_record `abstractmethod` ¶

insert_record(record: Record) -> RecordID

Upsert a record into the database.

PARAMETER	DESCRIPTION
`record`	The record to insert or update. TYPE: `Record`

RETURNS	DESCRIPTION
`RecordID`	The id of the given record.

batch_insert_record `abstractmethod` ¶

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

Upsert a batch of records into the database.

PARAMETER	DESCRIPTION
`records`	The records to insert or update. TYPE: `List[Record]`

RETURNS	DESCRIPTION
`List[RecordID]`	The ids of the given records.

insert_app `abstractmethod` ¶

insert_app(app: AppDefinition) -> AppID

Upsert an app into the database.

PARAMETER	DESCRIPTION
`app`	The app to insert or update. Note that only the AppDefinition parts are serialized hence the type hint. TYPE: `AppDefinition`

RETURNS	DESCRIPTION
`AppID`	The id of the given app.

delete_app `abstractmethod` ¶

delete_app(app_id: AppID) -> None

Delete an app from the database.

PARAMETER	DESCRIPTION
`app_id`	The id of the app to delete. TYPE: `AppID`

insert_feedback_definition `abstractmethod` ¶

insert_feedback_definition(
    feedback_definition: FeedbackDefinition,
) -> FeedbackDefinitionID

Upsert a feedback_definition into the database.

PARAMETER	DESCRIPTION
`feedback_definition`	The feedback definition to insert or update. Note that only the FeedbackDefinition parts are serialized hence the type hint. TYPE: `FeedbackDefinition`

RETURNS	DESCRIPTION
`FeedbackDefinitionID`	The id of the given feedback definition.

get_feedback_defs `abstractmethod` ¶

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

Retrieve feedback definitions from the database.

PARAMETER	DESCRIPTION
`feedback_definition_id`	if provided, only the feedback definition with the given id is returned. Otherwise, all feedback definitions are returned. TYPE: `Optional[FeedbackDefinitionID]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the feedback definitions.

insert_feedback `abstractmethod` ¶

insert_feedback(
    feedback_result: FeedbackResult,
) -> FeedbackResultID

Upsert a feedback_result into the the database.

PARAMETER	DESCRIPTION
`feedback_result`	The feedback result to insert or update. TYPE: `FeedbackResult`

RETURNS	DESCRIPTION
`FeedbackResultID`	The id of the given feedback result.

batch_insert_feedback `abstractmethod` ¶

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

Upsert a batch of feedback results into the database.

PARAMETER	DESCRIPTION
`feedback_results`	The feedback results to insert or update. TYPE: `List[FeedbackResult]`

RETURNS	DESCRIPTION
`List[FeedbackResultID]`	The ids of the given feedback results.

get_feedback `abstractmethod` ¶

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] = None,
    run_location: Optional[FeedbackRunLocation] = None,
) -> DataFrame

Get feedback results matching a set of optional criteria:

PARAMETER	DESCRIPTION
`record_id`	Get only the feedback for the given record id. TYPE: `Optional[RecordID]` DEFAULT: `None`
`feedback_result_id`	Get only the feedback for the given feedback result id. TYPE: `Optional[FeedbackResultID]` DEFAULT: `None`
`feedback_definition_id`	Get only the feedback for the given feedback definition id. TYPE: `Optional[FeedbackDefinitionID]` DEFAULT: `None`
`status`	Get only the feedback with the given status. If a sequence of statuses is given, all feedback with any of the given statuses are returned. TYPE: `Optional[Union[FeedbackResultStatus, Sequence[FeedbackResultStatus]]]` DEFAULT: `None`
`last_ts_before`	get only results with `last_ts` before the given datetime. TYPE: `Optional[datetime]` DEFAULT: `None`
`offset`	index of the first row to return. TYPE: `Optional[int]` DEFAULT: `None`
`limit`	limit the number of rows returned. TYPE: `Optional[int]` DEFAULT: `None`
`shuffle`	shuffle the rows before returning them. TYPE: `Optional[bool]` DEFAULT: `None`
`run_location`	Only get feedback functions with this run_location. TYPE: `Optional[FeedbackRunLocation]` DEFAULT: `None`

get_feedback_count_by_status `abstractmethod` ¶

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]

Get count of feedback results matching a set of optional criteria grouped by their status.

See get_feedback for the meaning of the the arguments.

RETURNS	DESCRIPTION
`Dict[FeedbackResultStatus, int]`	A mapping of status to the count of feedback results of that status that match the given filters.

get_app `abstractmethod` ¶

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

Get the app with the given id from the database.

RETURNS	DESCRIPTION
`Optional[JSONized]`	The jsonized version of the app with the given id. Deserialization can be done with App.model_validate.

get_apps `abstractmethod` ¶

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

Get all apps.

update_app_metadata ¶

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

Update the metadata of an app.

get_records_and_feedback `abstractmethod` ¶

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

Get records from the database.

PARAMETER	DESCRIPTION
`app_ids`	If given, retrieve only the records for the given apps. Otherwise all apps are retrieved. TYPE: `Optional[List[AppID]]` DEFAULT: `None`
`offset`	Database row offset. TYPE: `Optional[int]` DEFAULT: `None`
`limit`	Limit on rows (records) returned. TYPE: `Optional[int]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A DataFrame with the records.
`Sequence[str]`	A list of column names that contain feedback results.

insert_ground_truth `abstractmethod` ¶

insert_ground_truth(
    ground_truth: GroundTruth,
) -> GroundTruthID

Insert a ground truth entry into the database. The ground truth id is generated based on the ground truth content, so re-inserting is idempotent.

PARAMETER	DESCRIPTION
`ground_truth`	The ground truth entry to insert. TYPE: `GroundTruth`

RETURNS	DESCRIPTION
`GroundTruthID`	The id of the given ground truth entry.

batch_insert_ground_truth `abstractmethod` ¶

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

Insert a batch of ground truth entries into the database.

PARAMETER	DESCRIPTION
`ground_truths`	The ground truth entries to insert. TYPE: `List[GroundTruth]`

RETURNS	DESCRIPTION
`List[GroundTruthID]`	The ids of the given ground truth entries.

get_ground_truth `abstractmethod` ¶

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

Get the ground truth with the given id from the database.

get_ground_truths_by_dataset `abstractmethod` ¶

get_ground_truths_by_dataset(
    dataset_name: str,
) -> DataFrame

Get all ground truths from the database from a particular dataset's name.

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the ground truths.

insert_dataset `abstractmethod` ¶

insert_dataset(dataset: Dataset) -> DatasetID

Insert a dataset into the database. The dataset id is generated based on the dataset content, so re-inserting is idempotent.

PARAMETER	DESCRIPTION
`dataset`	The dataset to insert. TYPE: `Dataset`

RETURNS	DESCRIPTION
`DatasetID`	The id of the given dataset.

get_datasets `abstractmethod` ¶

get_datasets() -> DataFrame

Get all datasets from the database.

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the datasets.

trulens.core.database.base¶