❄️ Logging in Snowflake

Snowflake’s fully managed data warehouse provides automatic provisioning, availability, tuning, data protection and more—across clouds and regions—for an unlimited number of users and jobs.

TruLens can write and read from a Snowflake database using a SQLAlchemy connection. This allows you to read, write, persist and share TruLens logs in a Snowflake database.

Here is a guide to logging in Snowflake.

Install the TruLens Snowflake Connector

Install using pip

pip install trulens-connectors-snowflake

Connect TruLens to the Snowflake database

Connecting TruLens to a Snowflake database for logging traces and evaluations only requires passing in Snowflake connection parameters.

Connect TruLens to your Snowflake database

from trulens.core import TruSession
from trulens.connectors.snowflake import SnowflakeConnector
conn = SnowflakeConnector(
    account="<account>",
    user="<user>",
    password="<password>",
    database_name="<database>",
    schema_name="<schema>",
    warehouse="<warehouse>",
    role="<role>",
)
session = TruSession(connector=conn)

Once you've instantiated the TruSession object with your Snowflake connection, all TruLens traces and evaluations will logged to Snowflake.

Connect TruLens to the Snowflake database using an engine

In some cases such as when using key-pair authentication, the SQL-alchemy URL does not support the credentials required. In this case, you can instead create and pass a database engine.

When the database engine is created, the private key is then passed through the connection_args.

Connect TruLens to Snowflake with a database engine

from trulens.core import Tru
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization

load_dotenv()

with open("rsa_key.p8", "rb") as key:
    p_key= serialization.load_pem_private_key(
        key.read(),
        password=None,
        backend=default_backend()
    )

pkb = p_key.private_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption())

engine = create_engine(URL(
account=os.environ["SNOWFLAKE_ACCOUNT"],
warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
database=os.environ["SNOWFLAKE_DATABASE"],
schema=os.environ["SNOWFLAKE_SCHEMA"],
user=os.environ["SNOWFLAKE_USER"],),
connect_args={
        'private_key': pkb,
        },
)

from trulens.core import TruSession

session = TruSession(
    database_engine = engine
)