Teradata SQL Dialect for SQLAlchemy

20.0.0.9 · active · verified Thu Apr 09

teradatasqlalchemy provides a SQL dialect for SQLAlchemy, enabling Python applications to connect to and interact with the Teradata Database. It offers an abstraction layer over the Teradata DBAPI, allowing users to leverage SQLAlchemy's ORM or Core features. The library supports 64-bit Python 3.4 and later. The current version is 20.0.0.9, released on December 15, 2025, with frequent updates to support new Teradata datatypes and features.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to establish a connection to a Teradata database using teradatasqlalchemy and SQLAlchemy's `create_engine` function. It uses environment variables for secure credential management and executes a basic query to verify the connection. The `teradatasql://` dialect is recommended for direct connections without ODBC.

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.")

view raw JSON →