trulens.connectors.snowflake.sqlalchemy_db¶

trulens.connectors.snowflake.sqlalchemy_db ¶

Snowflake-specific SQLAlchemyDB subclass.

Encapsulates dialect-specific behavior that previously lived in trulens-core:

AUTOCOMMIT isolation_level workaround for snowflake-sqlalchemy >= 1.7.2.
NULL-result feedback insert workaround for the Snowflake stored-procedure connector.
JSON path extraction via json_extract_path_text with quoted paths.
Latency expression via timestampdiff.
Alembic DefaultImpl registration for the snowflake dialect.

Classes¶

SnowflakeImpl ¶

Bases: DefaultImpl

Alembic DDL impl marker for the snowflake dialect.

SnowflakeSQLAlchemyDB ¶

Bases: SQLAlchemyDB

SQLAlchemyDB subclass that hosts all Snowflake-dialect-specific overrides.

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¶

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.

str ¶

__str__() -> str

Relatively concise identifier string for this instance.

reset_database ¶

reset_database()

See DB.reset_database.

migrate_database ¶

migrate_database(prior_prefix: Optional[str] = None)

See DB.migrate_database.

check_db_revision ¶

check_db_revision()

See DB.check_db_revision.

insert_record ¶

insert_record(record: Record) -> RecordID

See DB.insert_record.

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 ¶

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_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_app ¶

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

See DB.get_app.

get_apps ¶

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

See DB.get_apps.

update_app_metadata ¶

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

See DB.update_app_metadata.

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, Sequence[str]]

See DB.get_records_and_feedback.

get_leaderboard_aggregates ¶

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

See DB.get_leaderboard_aggregates.

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.

insert_event ¶

insert_event(event: Event) -> EventID

See DB.insert_event.

get_events ¶

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

See DB.get_events.

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.

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.

insert_events ¶

insert_events(events: List[Event]) -> List[EventID]

Insert a batch of events into the database.

trulens.connectors.snowflake.sqlalchemy_db¶