I am using below code to get create table queries for postgresql database using psycopg2 connection. How can I get the same using SQLAlchemy?
def get_create_table_queries():
connection = connect_to_database()
if connection:
cursor = connection.cursor()
cursor.execute(
"""
SELECT
'CREATE TABLE ' || table_name || ' (' || STRING_AGG(column_definition, ', ') || ');'
FROM (
SELECT
table_name,
column_name || ' ' || data_type ||
CASE
WHEN character_maximum_length IS NOT NULL THEN '(' || character_maximum_length || ')'
ELSE ''
END AS column_definition
FROM information_schema.columns
WHERE table_schema = 'public'
) AS table_columns
GROUP BY table_name;
"""
)
create_table_queries = cursor.fetchall()
connection.close()
return create_table_queries
else:
return []
def get_create_table_queries():
try:
# Step 2: Create a SQLAlchemy Engine
engine = create_engine(os.getenv("DATABASE_CONNECTION_URL"))
# Step 3: Create a MetaData object without a bind
metadata = MetaData()
reflect = metadata.reflect(bind=engine)
create_table_queries = []
# Step 4: Use the inspect module to get table names
inspector = inspect(engine)
table_names = inspector.get_table_names()
# Step 5: Reflect individual tables with a specific bind and retrieve CREATE TABLE queries
for table_name in table_names:
table = Table(table_name, metadata, autoload_with=engine)
create_table_query = text(table.schema())
create_table_queries.append(str(create_table_query))
return create_table_queries
except Exception as e:
print(f"Error getting CREATE TABLE queries: {e}")
return []
I tried above but getting errors.
Error getting CREATE TABLE queries: 'NoneType' object is not callabl