trulens.core.database.connector¶

trulens.core.database.connector ¶

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.

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,
    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`
`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`	DataFrame of records with their feedback results.
`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.

DefaultDBConnector ¶

Bases: DBConnector

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.

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.

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,
    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`
`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`	DataFrame of records with their feedback results.
`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.

init ¶

__init__(
    database: Optional[DB] = None,
    database_url: Optional[str] = None,
    database_engine: Optional[Engine] = None,
    database_redact_keys: bool = False,
    database_prefix: Optional[str] = None,
    database_args: Optional[Dict[str, Any]] = None,
    database_check_revision: bool = True,
)

Create a default DB connector backed by a database.

To connect to an existing database, one of database, database_url, or database_engine must be provided.

PARAMETER	DESCRIPTION
`database`	The database object to use. TYPE: `Optional[DB]` DEFAULT: `None`
`database_url`	The database URL to connect to. To connect to a local file-based SQLite database, use `sqlite:///path/to/database.db`. TYPE: `Optional[str]` DEFAULT: `None`
`database_engine`	The SQLAlchemy engine object to use. TYPE: `Optional[Engine]` DEFAULT: `None`
`database_redact_keys`	Whether to redact keys in the database. TYPE: `bool` DEFAULT: `False`
`database_prefix`	The database prefix to use to separate tables in the database. TYPE: `Optional[str]` DEFAULT: `None`
`database_args`	Additional arguments to pass to the database. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`database_check_revision`	Whether to compare the database revision with the expected TruLens revision. TYPE: `bool` DEFAULT: `True`

trulens.core.database.connector¶