Postgres 'virtual' persistence error when running get_table_info() for SQLDatabase #26437
Replies: 1 comment 2 replies
-
|
Hey there, @BrandenKeck! I'm here to help you with any bugs, questions, or contribution assistance you might need. Let's tackle this issue together! To resolve the Here is an example of how to use import os
from langchain_community.utilities import SQLDatabase
user = os.getenv("USER")
pwd = os.getenv("PWD")
db = SQLDatabase.from_uri(
f"denodo://{user}:{pwd}@my.url.com:0000/database",
include_tables=["my_table"],
lazy_table_reflection=True
)
# Now you can call get_table_info without encountering the CompileError
table_info = db.get_table_info()
print(table_info)By using |
Beta Was this translation helpful? Give feedback.
-
|
Thanks, dosubot, but the |
Beta Was this translation helpful? Give feedback.
-
|
To resolve the
Here is an example of how you might modify the from sqlalchemy import inspect, create_engine, MetaData
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.sql import text
from sqlalchemy.sql.ddl import CreateTable
from sqlalchemy.types import NullType
class SQLDatabase:
def __init__(self, engine, schema=None, metadata=None, lazy_table_reflection=False):
self._engine = engine
self._schema = schema
self._inspector = inspect(engine)
self._metadata = metadata or MetaData()
self.lazy_table_reflection = lazy_table_reflection
if not lazy_table_reflection:
self._metadata.reflect(bind=self._engine, schema=self._schema)
def get_table_info(self, table_names):
if self.lazy_table_reflection:
for table_name in table_names:
if table_name not in self._metadata.tables:
self._metadata.reflect(self._engine, only=[table_name], schema=self._schema)
else:
self._metadata.reflect(self._engine, schema=self._schema)
table_info = []
for table_name in table_names:
table = self._metadata.tables.get(table_name)
if table is not None:
# Handle computed columns
for column in table.columns:
if hasattr(column, 'computed') and column.computed is not None:
column.computed.persisted = True # Set persisted flag to True
create_table = str(CreateTable(table).compile(self._engine))
table_info.append(create_table)
return "\n".join(table_info)
@classmethod
def from_uri(cls, database_uri, **kwargs):
engine = create_engine(database_uri)
return cls(engine, **kwargs)
# Example usage
engine = create_engine("denodo://username:password@host:port/database")
db = SQLDatabase(engine, lazy_table_reflection=True)
print(db.get_table_info(["your_table_name"]))This code sets the |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
Checked other resources
Commit to Help
Example Code
Description
I am trying to use the SQLDatabase class to connect to a Denodo (https://www.denodo.com/en) database. The connection works using the above URI structure and I can query data using
db.run(...)just fine. My problem is that I would like to invoke the following pipeline:Which appears to depend on:
And, when this function is called, I get the following Error:
Any help with this would be extremely appreciated. The full traceback is as follows:
System Info
langchain==0.2.15
langchain-community==0.2.15
langchain-core==0.2.37
langchain-text-splitters==0.2.2
Ubuntu 22.04
Py3.10.14
Beta Was this translation helpful? Give feedback.
All reactions