SQL

SQL#

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 RowResultdataclass s.

  • execute_query_row() executes a query and returns a raw result from SQL.

  • get_table_schema() returns a table schema.

SQL Drivers#

SQL#

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#

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#

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;")
Contents