SQLAlchemy Exasol Dialect

6.1.1 · active · verified Thu Apr 16

SQLAlchemy-Exasol is a dialect for SQLAlchemy that allows Python applications to connect and interact with an Exasol database. It provides support for both WebSocket and ODBC-based connections to Exasol. The current version is 6.1.1, with releases typically occurring every 1-3 months to introduce new features, fix bugs, and maintain compatibility with Python and SQLAlchemy updates.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to establish a connection to an Exasol database using `sqlalchemy-exasol` with the default WebSocket driver. It configures connection parameters from environment variables (or defaults), creates an engine, tests the connection with a simple query, and shows how to create a schema and a table.

from sqlalchemy import create_engine, text
import os

# Configure connection details from environment variables or provide defaults
EXASOL_USER = os.environ.get("EXASOL_USER", "sys")
EXASOL_PASSWORD = os.environ.get("EXASOL_PASSWORD", "exasol")
EXASOL_HOST = os.environ.get("EXASOL_HOST", "localhost")
EXASOL_PORT = os.environ.get("EXASOL_PORT", "8563")
EXASOL_SCHEMA = os.environ.get("EXASOL_SCHEMA", "SYS")

# Construct the connection string using the default websocket driver
# For ODBC, use 'exasol+pyodbc://'
connection_string = f"exasol+websocket://{EXASOL_USER}:{EXASOL_PASSWORD}@{EXASOL_HOST}:{EXASOL_PORT}/{EXASOL_SCHEMA}"

try:
    # Create the engine
    engine = create_engine(connection_string)

    # Test the connection and execute a simple query
    with engine.connect() as connection:
        result = connection.execute(text("SELECT 1 FROM DUAL")).scalar()
        print(f"Successfully connected to Exasol. Query result: {result}")

    # Example: Create a table
    with engine.connect() as connection:
        connection.execute(text("CREATE SCHEMA IF NOT EXISTS MY_SCHEMA"))
        connection.execute(text("CREATE TABLE IF NOT EXISTS MY_SCHEMA.test_table (id INT, name VARCHAR(255))"))
        connection.commit()
        print("Table MY_SCHEMA.test_table created or already exists.")

except Exception as e:
    print(f"An error occurred: {e}")

view raw JSON →