SQL

Adapted from the Griptape AI Framework documentation.

Overview

SQL drivers can be used to make SQL queries and load table schemas. They are used by the SqlLoader to process data. All loaders implement the following methods:

execute_query() executes a query and returns RowResults.
execute_query_row() executes a query and returns a raw result from SQL.
get_table_schema() returns a table schema.

SQL Drivers

SQL

Info

This driver requires the drivers-sql extra.

Note that you may need to install the appropriate database driver for your SQL database. For example, to use the psycopg2 driver for PostgreSQL, you can install it with pip install psycopg2-binary.

This is a basic SQL loader based on SQLAlchemy 2.0. Here is an example of how to use it:

from griptape.drivers.sql.sql_driver import SqlDriver

driver = SqlDriver(engine_url="sqlite:///:memory:")

driver.execute_query("select 'foo', 'bar';")

Amazon Redshift

Info

This driver requires the drivers-sql-amazon-redshift extra.

This is a SQL driver for interacting with the Amazon Redshift Data API to execute statements. Here is an example of how to use it for Redshift Serverless:

import os

import boto3

from griptape.drivers.sql.amazon_redshift import AmazonRedshiftSqlDriver

session = boto3.Session()

driver = AmazonRedshiftSqlDriver(
    database=os.environ["REDSHIFT_DATABASE"],
    session=session,
    cluster_identifier=os.environ["REDSHIFT_CLUSTER_IDENTIFIER"],
)

driver.execute_query("select * from people;")

Snowflake

Info

This driver requires the drivers-sql-snowflake extra.

This is a SQL driver based on the Snowflake SQLAlchemy Toolkit which runs on top of the Snowflake Connector for Python. Here is an example of how to use it:

import os

import snowflake.connector
from snowflake.connector import SnowflakeConnection

from griptape.drivers.sql.snowflake import SnowflakeSqlDriver


def get_snowflake_connection() -> SnowflakeConnection:
    return snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
        database=os.environ["SNOWFLAKE_DATABASE"],
        schema=os.environ["SNOWFLAKE_SCHEMA"],
        warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
    )


driver = SnowflakeSqlDriver(get_connection=get_snowflake_connection)

driver.execute_query("select * from people;")