trulens.connectors.snowflake.snowflake_event_table_db¶

trulens.connectors.snowflake.snowflake_event_table_db ¶

Classes¶

SnowflakeEventTableDB ¶

Bases: DB

Connector to the account level event table in Snowflake.

Attributes¶

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

redact_keys: bool = DEFAULT_DATABASE_REDACT_KEYS

Redact secrets before writing out data.

Functions¶

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,
    matched_record_ids: Optional[List[RecordID]] = None,
    include_conversation_context: bool = False,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
    offset: Optional[int] = None,
    limit: Optional[int] = None,
) -> Tuple[DataFrame, Sequence[str]]

See DB.get_records_and_feedback.

get_events ¶

get_events(
    app_name: Optional[AppName] = None,
    app_version: Optional[AppVersion] = None,
    run_name: Optional[RunName] = None,
    record_ids: Optional[List[RecordID]] = 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[AppName]` DEFAULT: `None`
`app_version`	The app version to filter events by. TYPE: `Optional[AppVersion]` DEFAULT: `None`
`run_name`	The run name to filter events by. TYPE: `Optional[RunName]` DEFAULT: `None`
`record_ids`	The record ids to filter events by. TYPE: `Optional[List[RecordID]]` 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.

get_apps ¶

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

See DB.get_apps.

get_feedback_defs ¶

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

See DB.get_feedback_defs.

repr ¶

__repr__() -> str

Safe repr that handles circular references.

Pydantic's default __repr__ does not guard against circular references among model instances, which leads to RecursionError (see GitHub issue #1862). This override uses the same formatted_objects context-variable that __rich_repr__ uses so that already-visited objects are replaced with a short placeholder instead of recursing infinitely.

__rich_repr__ ¶

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

update_app_metadata ¶

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

Update the metadata of an app.

get_conversations ¶

get_conversations(
    app_id: Optional[AppID] = None,
) -> Dict[ConversationID, Conversation]

Get conversations reconstructed from recorded spans.

get_records_by_conversation ¶

get_records_by_conversation(
    conversation_id: ConversationID,
    app_id: Optional[AppID] = None,
) -> List[Record]

Get a conversation's records ordered chronologically.

get_leaderboard_aggregates ¶

get_leaderboard_aggregates(
    app_name: Optional[AppName] = None,
    app_versions: Optional[List[AppVersion]] = None,
) -> Tuple[DataFrame, List[str]]

Get pre-aggregated leaderboard data via SQL GROUP BY.

Returns per-app-version aggregates (record count, avg latency, total cost, avg feedback scores) without fetching individual records.

PARAMETER	DESCRIPTION
`app_name`	App name to filter by. TYPE: `Optional[AppName]` DEFAULT: `None`
`app_versions`	List of app versions to filter by. TYPE: `Optional[List[AppVersion]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tuple[DataFrame, List[str]]`	A tuple of (aggregated dataframe, feedback column names).

get_feedback_score_trends ¶

get_feedback_score_trends(
    app_name: Optional[AppName] = None,
    app_versions: Optional[List[AppVersion]] = None,
    bucket: str = "day",
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
) -> DataFrame

Get time-bucketed feedback score statistics.

PARAMETER	DESCRIPTION
`app_name`	App name to filter by. TYPE: `Optional[AppName]` DEFAULT: `None`
`app_versions`	App versions to filter by. TYPE: `Optional[List[AppVersion]]` DEFAULT: `None`
`bucket`	Time bucket, either `day` or `week`. TYPE: `str` DEFAULT: `'day'`
`start_time`	Inclusive event timestamp lower bound. TYPE: `Optional[datetime]` DEFAULT: `None`
`end_time`	Exclusive event timestamp upper bound. TYPE: `Optional[datetime]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	Feedback score counts, means, and 95% confidence bounds grouped by
`DataFrame`	app version, metric, and time bucket.

get_app_metric_trends ¶

get_app_metric_trends(
    app_name: Optional[AppName] = None,
    app_versions: Optional[List[AppVersion]] = None,
    bucket: str = "day",
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
) -> DataFrame

Get time-bucketed application latency and cost statistics.

get_eval_cost_trends ¶

get_eval_cost_trends(
    app_name: Optional[AppName] = None,
    app_versions: Optional[List[AppVersion]] = None,
    bucket: str = "day",
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
) -> DataFrame

Get time-bucketed observed evaluation cost statistics.

get_eval_drilldown_record_ids ¶

get_eval_drilldown_record_ids(
    app_name: AppName,
    app_versions: List[AppVersion],
    metric_kind: str,
    metric_name: str,
    start_time: datetime,
    end_time: datetime,
    currency: Optional[str] = None,
) -> List[RecordID]

Get record IDs for an evaluation-time chart bucket.

insert_prompt ¶

insert_prompt(prompt: Prompt) -> PromptID

Insert a prompt, or update the metadata of an existing one.

The prompt id is derived from the slug, so re-inserting the same slug updates the name, description, and tags in place.

PARAMETER	DESCRIPTION
`prompt`	The prompt to insert. TYPE: `Prompt`

RETURNS	DESCRIPTION
`PromptID`	The id of the given prompt.

RAISES	DESCRIPTION
`ValueError`	If a prompt with the same slug exists with a different prompt type.

get_prompt ¶

get_prompt(
    prompt_id: Optional[PromptID] = None,
    slug: Optional[str] = None,
) -> Optional[Prompt]

Get one prompt by id or by slug.

PARAMETER	DESCRIPTION
`prompt_id`	The id to look up. TYPE: `Optional[PromptID]` DEFAULT: `None`
`slug`	The slug to look up. Ignored when `prompt_id` is given. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Optional[Prompt]`	The prompt, or None when it does not exist.

get_prompts ¶

get_prompts() -> DataFrame

Get all prompts from the database.

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the prompts.

insert_prompt_version ¶

insert_prompt_version(
    version: PromptVersion, move_latest: bool = True
) -> PromptVersionID

Insert an immutable prompt version.

Inserting a version that already exists is a no-op, which makes creation idempotent.

PARAMETER	DESCRIPTION
`version`	The version to insert. TYPE: `PromptVersion`
`move_latest`	Whether to move the `latest` label onto this version in the same transaction. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`PromptVersionID`	The id of the given version.

get_prompt_version ¶

get_prompt_version(
    version_id: PromptVersionID,
) -> Optional[PromptVersion]

Get one exact prompt version.

PARAMETER	DESCRIPTION
`version_id`	The version to look up. TYPE: `PromptVersionID`

RETURNS	DESCRIPTION
`Optional[PromptVersion]`	The version, or None when it does not exist.

get_prompt_versions ¶

get_prompt_versions(prompt_id: PromptID) -> DataFrame

Get every version of a prompt, oldest first.

PARAMETER	DESCRIPTION
`prompt_id`	The prompt whose versions to list. TYPE: `PromptID`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the versions.

set_prompt_label ¶

set_prompt_label(
    prompt_id: PromptID,
    label: str,
    version_id: PromptVersionID,
    moved_by: Optional[str] = None,
) -> PromptLabel

Point a label at one exact version and record the movement.

PARAMETER	DESCRIPTION
`prompt_id`	The prompt whose label to move. TYPE: `PromptID`
`label`	The label name. TYPE: `str`
`version_id`	The version to point at. TYPE: `PromptVersionID`
`moved_by`	Caller label written to the history entry. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`PromptLabel`	The resulting label pointer.

RAISES	DESCRIPTION
`ValueError`	If the version does not belong to the prompt.

get_prompt_label ¶

get_prompt_label(
    prompt_id: PromptID, label: str
) -> Optional[PromptLabel]

Get the current pointer for one label.

PARAMETER	DESCRIPTION
`prompt_id`	The prompt to look in. TYPE: `PromptID`
`label`	The label name. TYPE: `str`

RETURNS	DESCRIPTION
`Optional[PromptLabel]`	The label pointer, or None when the label is unset.

get_prompt_labels ¶

get_prompt_labels(prompt_id: PromptID) -> DataFrame

Get every label of a prompt.

PARAMETER	DESCRIPTION
`prompt_id`	The prompt whose labels to list. TYPE: `PromptID`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the labels.

get_prompt_label_history ¶

get_prompt_label_history(
    prompt_id: PromptID, label: Optional[str] = None
) -> DataFrame

Get the append-only history of label movements, newest first.

PARAMETER	DESCRIPTION
`prompt_id`	The prompt whose history to read. TYPE: `PromptID`
`label`	Restrict to one label when given. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with the history entries.

extract_app_and_run_info `staticmethod` ¶

extract_app_and_run_info(
    attributes: Dict[str, Any],
    resource_attributes: Dict[str, Any],
) -> Tuple[str, str, str, str]

Get app info from attributes.

PARAMETER	DESCRIPTION
`attributes`	Span attributes of record root. TYPE: `Dict[str, Any]`
`resource_attributes`	Resource attributes of record root. TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`Tuple[str, str, str, str]`	Tuple of: app name, app version, app id, and run name.

trulens.connectors.snowflake.snowflake_event_table_db¶