teradatasqlalchemy

import os from sqlalchemy import create_engine, text # Get connection details from environment variables for security and flexibility td_host = os.environ.get("TD_HOST", "your_teradata_host") td_user = os.environ.get("TD_USER", "your_username") td_password = os.environ.get("TD_PASSWORD", "your_password") td_database = os.environ.get("TD_DATABASE", "your_database") # Optional, can be empty # Construct the connection string using the 'teradatasql' dialect # This dialect (teradatasql://) does NOT require an ODBC driver. # If td_database is provided, include it in the connection string. if td_database: connection_string = f"teradatasql://{td_user}:{td_password}@{td_host}/?database={td_database}" else: connection_string = f"teradatasql://{td_user}:{td_password}@{td_host}" print(f"Attempting to connect to Teradata at {td_host}...") try: # Create the SQLAlchemy engine engine = create_engine(connection_string) # Establish a connection and execute a simple query with engine.connect() as connection: # Example: Execute a simple SELECT 1 query to verify connection result = connection.execute(text("SELECT 1 AS test_column")).scalar() print(f"Connection successful! Query result: {result}") # Example: Execute a query to fetch data from a table # Make sure 'your_table' and 'your_column' exist in your Teradata database # try: # sample_data = connection.execute(text("SELECT TOP 5 column1 FROM your_table")).fetchall() # print("Sample data:", sample_data) # except Exception as query_e: # print(f"Error executing sample query: {query_e}") except Exception as e: print(f"Error connecting to Teradata: {e}") print("Please ensure environment variables (TD_HOST, TD_USER, TD_PASSWORD, TD_DATABASE) are set correctly ") print("and network connectivity to the Teradata host on port 1025 (default) is available.")