Database Agent with Snowflake connection does not infer database/schema from connection string

Short Course Q&A LangChain for LLM Application Development
1

I am trying to connect to Snowflake using the database agent SQLDatabase.from_uri as follows:

def generate_response(message):
    # Connect to the database
    dburi = f'snowflake://{USER}:{PASSWORD}@{ACCOUNT.lower()}/{DATABASE}/{SCHEMA}?warehouse=DBT_CLOUD'
    db = SQLDatabase.from_uri(dburi)

    # Create an instance of LLM
    llm = ChatOpenAI()

    # Create an SQLDatabaseChain using the ChatOpenAI model and the database
    db_chain = SQLDatabaseChain.from_llm(llm=llm, db=db)
    ai_response = db_chain.run(message)

    return ai_response

However the DB and Schema specified in the connection string is not being picked up and I get the following error:

  File "/skynet/venv/lib/python3.9/site-packages/snowflake/sqlalchemy/snowdialect.py", line 486, in _get_schema_columns
    full_schema_name = self._denormalize_quote_join(current_database, schema)
  File "/skynet/venv/lib/python3.9/site-packages/snowflake/sqlalchemy/snowdialect.py", line 289, in _denormalize_quote_join
    split_idents = reduce(
TypeError: reduce() of empty sequence with no initial value

Looking through the stacktrace, the issue stems from Database and Schema not being specifed in the SQL Alchemy engine.

2

The issue was the role I was using did not have the proper permissions to the database / schema, applying the proper grants to the user resolved this issue.

For anyone looking in the future, :wave: I used the following validate.py to trouble shoot each component:

#!/usr/bin/env python
import os
from sqlalchemy import create_engine
from dotenv import load_dotenv
import logging

# load envars
load_dotenv()



USER = os.getenv('SNOWFLAKE_USER')
PASSWORD = os.getenv('SNOWFLAKE_PASSWORD')
ACCOUNT = os.getenv('SNOWFLAKE_ACCOUNT')
DATABASE = os.getenv('SNOWFLAKE_DATABASE')
SCHEMA = os.getenv('SNOWFLAKE_SCHEMA')

engine = create_engine(
    f'snowflake://{USER}:{PASSWORD}@{ACCOUNT}/{DATABASE}/{SCHEMA}?warehouse=DBT_CLOUD&role=ACCOUNTADMIN'
)

try:
    connection = engine.connect()

    results = connection.execute('select current_user()').fetchone()
    print(results[0])

    results = connection.execute('select current_role()').fetchone()
    print(results[0])

    res = connection.execute(
            "select current_database();"
        ).fetchone()
    print(res[0])
finally:
    connection.close()
    engine.dispose()