trulens.core.database.connector.base¶

trulens.core.database.connector.base ¶

Classes¶

DBConnector ¶

Bases: ABC, WithIdentString

Base class for DB connector implementations.

Attributes¶

RECORDS_BATCH_TIMEOUT_IN_SEC `class-attribute` `instance-attribute` ¶

RECORDS_BATCH_TIMEOUT_IN_SEC: int = 10

Time to wait before inserting a batch of records into the database.

db `abstractmethod` `property` ¶

db: DB

Get the database instance.

Functions¶

reset_database ¶

reset_database()

Reset the database. Clears all tables.

See DB.reset_database.

migrate_database ¶

migrate_database(**kwargs: Any)

Migrates the database.

This should be run whenever there are breaking changes in a database created with an older version of trulens.

PARAMETER	DESCRIPTION
`**kwargs`	Keyword arguments to pass to migrate_database of the current database. TYPE: `Any` DEFAULT: `{}`

See DB.migrate_database.

add_record ¶

add_record(
    record: Optional[Record] = None, **kwargs
) -> RecordID

Add a record to the database.

PARAMETER	DESCRIPTION
`record`	The record to add. TYPE: `Optional[Record]` DEFAULT: `None`
`**kwargs`	Record fields to add to the given record or a new record if no `record` provided. DEFAULT: `{}`

RETURNS	DESCRIPTION
`RecordID`	Unique record identifier str .

add_record_nowait ¶

add_record_nowait(record: Record) -> None

Add a record to the queue to be inserted in the next batch.

add_app ¶

add_app(app: AppDefinition) -> AppID

Add an app to the database and return its unique id.

PARAMETER	DESCRIPTION
`app`	The app to add to the database. TYPE: `AppDefinition`

RETURNS	DESCRIPTION
`AppID`	A unique app identifier str.

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`

add_feedback_definition ¶

add_feedback_definition(
    feedback_definition: FeedbackDefinition,
) -> FeedbackDefinitionID

Add a feedback definition to the database and return its unique id.

PARAMETER	DESCRIPTION
`feedback_definition`	The feedback definition to add to the database. TYPE: `FeedbackDefinition`

RETURNS	DESCRIPTION
`FeedbackDefinitionID`	A unique feedback definition identifier str.

add_feedback ¶

add_feedback(
    feedback_result_or_future: Optional[
        Union[FeedbackResult, Future[FeedbackResult]]
    ] = None,
    **kwargs: Any
) -> FeedbackResultID

Add a single feedback result or future to the database and return its unique id.

PARAMETER	DESCRIPTION
`feedback_result_or_future`	If a Future is given, call will wait for the result before adding it to the database. If `kwargs` are given and a FeedbackResult is also given, the `kwargs` will be used to update the FeedbackResult otherwise a new one will be created with `kwargs` as arguments to its constructor. TYPE: `Optional[Union[FeedbackResult, Future[FeedbackResult]]]` DEFAULT: `None`
`**kwargs`	Fields to add to the given feedback result or to create a new FeedbackResult with. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`FeedbackResultID`	A unique result identifier str.

add_feedbacks ¶

add_feedbacks(
    feedback_results: Iterable[
        Union[FeedbackResult, Future[FeedbackResult]]
    ]
) -> List[FeedbackResultID]

Add multiple feedback results to the database and return their unique ids.

TODO: This is slow and should be batched or otherwise optimized in the future.¶

PARAMETER	DESCRIPTION
`feedback_results`	An iterable with each iteration being a FeedbackResult or Future of the same. Each given future will be waited. TYPE: `Iterable[Union[FeedbackResult, Future[FeedbackResult]]]`

RETURNS	DESCRIPTION
`List[FeedbackResultID]`	List of unique result identifiers str in the same order as input `feedback_results`.

get_app ¶

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

Look up an app from the database.

This method produces the JSON-ized version of the app. It can be deserialized back into an AppDefinition with model_validate:

Example

from trulens.core.schema import app
app_json = session.get_app(app_id="Custom Application v1")
app = app.AppDefinition.model_validate(app_json)

Warning

Do not rely on deserializing into App as its implementations feature attributes not meant to be deserialized.

PARAMETER	DESCRIPTION
`app_id`	The unique identifier str of the app to look up. TYPE: `AppID`

RETURNS	DESCRIPTION
`Optional[JSONized[AppDefinition]]`	JSON-ized version of the app.

get_apps ¶

get_apps() -> List[JSONized[AppDefinition]]

Look up all apps from the database.

RETURNS	DESCRIPTION
`List[JSONized[AppDefinition]]`	A list of JSON-ized version of all apps in the database.

Warning

Same Deserialization caveats as get_app.

get_records_and_feedback ¶

get_records_and_feedback(
    app_ids: Optional[List[AppID]] = None,
    app_name: Optional[AppName] = None,
    app_version: Optional[AppVersion] = None,
    app_versions: Optional[List[AppVersion]] = None,
    run_name: Optional[RunName] = None,
    record_ids: Optional[List[RecordID]] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
) -> Tuple[DataFrame, List[str]]

Get records, their feedback results, and feedback names.

PARAMETER	DESCRIPTION
`app_ids`	A list of app ids to filter records by. If empty or not given, all apps' records will be returned. TYPE: `Optional[List[AppID]]` DEFAULT: `None`
`app_name`	A name of the app to filter records by. If given, only records for this app will be returned. TYPE: `Optional[AppName]` DEFAULT: `None`
`app_version`	A version of the app to filter records by. If given, only records for this app version will be returned. TYPE: `Optional[AppVersion]` DEFAULT: `None`
`app_versions`	A list of app versions to filter records by. If given, only records for these app versions will be returned. TYPE: `Optional[List[AppVersion]]` DEFAULT: `None`
`run_name`	A run name to filter records by. If given, only records for this run will be returned. TYPE: `Optional[RunName]` DEFAULT: `None`
`record_ids`	An optional list of record ids to filter records by. TYPE: `Optional[List[RecordID]]` DEFAULT: `None`
`offset`	Record row offset. TYPE: `Optional[int]` DEFAULT: `None`
`limit`	Limit on the number of records to return. TYPE: `Optional[int]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	Tuple of:
`List[str]`	DataFrame of records with their feedback results.
`Tuple[DataFrame, List[str]]`	List of feedback names that are columns in the DataFrame.

get_leaderboard ¶

get_leaderboard(
    app_ids: Optional[List[AppID]] = None,
    group_by_metadata_key: Optional[str] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
) -> DataFrame

Get a leaderboard for the given apps.

PARAMETER	DESCRIPTION
`app_ids`	A list of app ids to filter records by. If empty or not given, all apps will be included in leaderboard. TYPE: `Optional[List[AppID]]` DEFAULT: `None`
`group_by_metadata_key`	A key included in record metadata that you want to group results by. TYPE: `Optional[str]` DEFAULT: `None`
`limit`	Limit on the number of records to aggregate to produce the leaderboard. TYPE: `Optional[int]` DEFAULT: `None`
`offset`	Record row offset to select which records to use to aggregate the leaderboard. TYPE: `Optional[int]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	DataFrame of apps with their feedback results aggregated.
`DataFrame`	If group_by_metadata_key is provided, the DataFrame will be grouped by the specified key.

add_event ¶

add_event(event: Event)

Add an event to the database.

PARAMETER	DESCRIPTION
`event`	The event to add to the database. TYPE: `Event`

add_events ¶

add_events(events: List[Event])

Add multiple events to the database.

PARAMETER	DESCRIPTION
`events`	A list of events to add to the database. TYPE: `List[Event]`

get_events ¶

get_events(
    app_name: Optional[str] = None,
    app_version: Optional[str] = None,
    record_ids: Optional[List[str]] = None,
    start_time: Optional[datetime] = None,
) -> DataFrame

Get events from the database.

PARAMETER	DESCRIPTION
`app_name`	The app name to filter events by. TYPE: `Optional[str]` DEFAULT: `None`
`app_version`	The app version to filter events by. TYPE: `Optional[str]` DEFAULT: `None`
`record_ids`	The record ids to filter events by. TYPE: `Optional[List[str]]` DEFAULT: `None`
`start_time`	The minimum time to consider events from. TYPE: `Optional[datetime]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A pandas DataFrame of all relevant events.

trulens.core.database.connector.base¶