sqlalchemy-vertica 1.0.1

Features

• Full SQLAlchemy 2.0+ Architecture: Built on DefaultDialect with query caching (supports_statement_cache = True), 2.0 execution semantics, and parameter-bound reflection.

• First-Class Async Engine Support: Run queries asynchronously with create_async_engine() and AsyncSession via vertica+vertica_python_async:// without blocking the asyncio event loop.

• Alembic Migrations: Native VerticaImpl integration with transactional DDL, type synonym resolution, and index no-op handling (since Vertica utilizes projections).

• Multi-Driver Support: * vertica-python (Synchronous pure-Python DBAPI driver) * vertica-python-async (Asynchronous DBAPI adapter for non-blocking asyncio / FastAPI apps) * pyodbc (ODBC driver) * turbodbc (High-speed ODBC driver for Arrow / NumPy / Pandas data workflows)

• Rich Vertica Data Types: * Geospatial: GEOMETRY, GEOGRAPHY * Identifiers: native UUID * Large objects: LONG VARCHAR, LONG VARBINARY (up to 32MB) * Complex types: ARRAY, MAP, ROW (Vertica 10+) * Temporal: TIMESTAMPTZ, TIMETZ, INTERVAL

• Complete Reflection: Automatic introspection of schemas, tables, temp tables, views, view definitions, columns, primary keys, foreign keys, unique constraints, check constraints, table & column comments.

Installation

Install from PyPI with your desired driver extras:

# Pure Python sync driver (recommended for sync applications)
pip install "sqlalchemy-vertica[vertica-python]"

# Pure Python async driver (for AsyncEngine / FastAPI / asyncio)
pip install "sqlalchemy-vertica[asyncio]"

# ODBC drivers
pip install "sqlalchemy-vertica[pyodbc]"
pip install "sqlalchemy-vertica[turbodbc]"

# Alembic migrations support
pip install "sqlalchemy-vertica[alembic]"

# Install all drivers and tools
pip install "sqlalchemy-vertica[all]"

Connection Strings

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import create_async_engine

# 1. Async (for FastAPI / asyncio applications)
async_engine = create_async_engine(
    "vertica+vertica_python_async://user:pwd@host:5433/database?connection_timeout=10"
)

# 2. Sync vertica-python
engine = sa.create_engine(
    "vertica+vertica_python://user:pwd@host:5433/database?connection_timeout=10"
)

# 3. PyODBC with connection string
engine_pyodbc = sa.create_engine(
    "vertica+pyodbc:///?odbc_connect=DSN%3DVerticaDSN"
)

# 4. Turbodbc with DSN
engine_turbodbc = sa.create_engine(
    "vertica+turbodbc:///?DSN=VerticaDSN"
)

Quick Start

from sqlalchemy import create_engine, text

engine = create_engine("vertica+vertica_python://user:pwd@localhost:5433/mydb")

with engine.connect() as conn:
    result = conn.execute(text("SELECT version()"))
    print(result.scalar())

# Transaction block
with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO my_table (name) VALUES (:name)"),
        {"name": "Alice"}
    )

import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

async def main():
    engine = create_async_engine(
        "vertica+vertica_python_async://user:pwd@localhost:5433/mydb",
        pool_size=10,
    )

    async with engine.connect() as conn:
        result = await conn.execute(text("SELECT 1"))
        print(result.scalar())

    # Using AsyncSession
    session_factory = async_sessionmaker(engine, class_=AsyncSession)
    async with session_factory() as session:
        result = await session.execute(text("SELECT COUNT(*) FROM my_table"))
        print("Count:", result.scalar())

    await engine.dispose()

asyncio.run(main())

Alembic Migrations

In your Alembic env.py, simply import sqlalchemy_vertica:

import sqlalchemy_vertica  # Registers VerticaImpl plugin automatically
from alembic import context

# configure context
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    transactional_ddl=True,
)

Vertica does not support traditional B-tree indexes (it utilizes projections). sqlalchemy-vertica treats index creation/dropping as safe no-ops in migrations to ensure multi-database migration scripts run seamlessly.

Custom Data Types

from sqlalchemy import Column, Integer, Table, MetaData
from sqlalchemy_vertica import (
    GEOMETRY,
    GEOGRAPHY,
    UUID,
    LONG_VARCHAR,
    ARRAY,
    MAP,
    ROW,
    TIMESTAMPTZ,
)

metadata = MetaData()

places = Table(
    "places",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("guid", UUID, nullable=False),
    Column("description", LONG_VARCHAR),
    Column("location", GEOMETRY(srid=4326)),
    Column("tags", ARRAY(LONG_VARCHAR)),
    Column("metadata", MAP(LONG_VARCHAR, LONG_VARCHAR)),
    Column("created_at", TIMESTAMPTZ),
)

Testing & Coverage

Run the automated test suite with pytest and pytest-cov:

pytest -v --cov=sqlalchemy_vertica --cov-report=term-missing

Support

If you find this project helpful and want to support its maintenance and development, you can buy me a coffee:

License

MIT License. See LICENSE for details.