Apache Airflow ODBC Provider

Warnings

• breaking The `driver` parameter for `OdbcHook` must now be passed directly via the hook constructor or `hook_params` in SQL Operators, not through the connection's `extra` field, due to a security vulnerability fix.
  Fix: Update usage to pass `driver` as a keyword argument to `OdbcHook` or within the `hook_params` dictionary for SQL Operators. Example `extra` for connection: `{"autocommit": true}` and `driver='{ODBC Driver 18 for SQL Server}'` passed separately to the hook.
• breaking This provider version has increased minimum Airflow requirements. For example, provider `3.0.0` requires Airflow `2.2+` and provider `4.12.1` requires Airflow `2.11.0+`. Older provider versions had lower Airflow requirements.
  Fix: Ensure your Airflow core installation meets the minimum version requirement for the installed provider. Refer to the provider's changelog for specific version compatibility.
• breaking When passing keyword arguments to the ODBC connection via the `connect_kwargs` key in the connection's `extra` field (e.g., `autocommit`, `ansi`), values must now be booleans (e.g., `true`, `false`) instead of strings (`"true"`, `"false"`).
  Fix: Update your Airflow Connection's `extra` field to use boolean literals for `connect_kwargs` values: `"connect_kwargs": {"autocommit": true, "ansi": false}`.
• gotcha The `apache-airflow-providers-odbc` package and its `pyodbc` dependency require system-level ODBC drivers to be installed on your Airflow worker machines. These are not installed by `pip`.
  Fix: Manually install the appropriate ODBC driver for your database (e.g., Microsoft ODBC Driver for SQL Server) on the operating system where Airflow workers run.
• gotcha Enabling the `allow_driver_in_extra` configuration (via `AIRFLOW__PROVIDERS_ODBC__ALLOW_DRIVER_IN_EXTRA=true` environment variable or `airflow.cfg`) is a security risk. It allows users to specify arbitrary ODBC drivers via the Airflow Connection's `extra` field, potentially leading to privilege escalation or arbitrary code execution if untrusted users can modify connections.
  Fix: Avoid setting `allow_driver_in_extra` to `True` unless you fully trust all users who can modify Airflow connections. Ensure drivers are only specified directly in DAG code or via the hook constructor.
• gotcha Some users have reported `UnicodeDecodeError` (e.g., `'utf-8' codec can't decode byte 0x91`) when using specific ODBC drivers (like Netezza) with certain Airflow and provider versions.
  Fix: If encountering encoding errors, ensure the correct character set is configured in your ODBC driver and/or connection string. Consult the specific database and ODBC driver documentation, or consider upgrading/downgrading provider/Airflow versions if a known fix exists.

Install

• pip install apache-airflow-providers-odbc Install only ODBC provider
• pip install apache-airflow[odbc] Install Airflow with ODBC provider (includes pyodbc)

Imports

• OdbcHook

  from airflow.providers.odbc.hooks.odbc import OdbcHook

• OdbcOperator

  from airflow.providers.odbc.operators.odbc import OdbcOperator

• SQLExecuteQueryOperator

  from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

  As of provider version 3.1.0, SQL operators like SQLExecuteQueryOperator were moved to the common-sql provider.

Quickstart

This quickstart demonstrates a basic Airflow DAG using the `SQLExecuteQueryOperator` to interact with an ODBC database. It assumes an Airflow ODBC connection (`my_odbc_conn`) is configured in the UI or via environment variables, including the necessary ODBC driver. Remember to install system-level ODBC drivers for your specific database.

from __future__ import annotations

import pendulum

from airflow.models.dag import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Ensure you have an Airflow Connection named 'my_odbc_conn'
# type: ODBC
# host: your_odbc_server
# login: your_username
# password: your_password
# extra: {"driver": "{ODBC Driver 18 for SQL Server}", "autocommit": true}

with DAG(
    dag_id="odbc_example_dag",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    schedule=None,
    catchup=False,
    tags=["odbc", "example"],
) as dag:
    create_table = SQLExecuteQueryOperator(
        task_id="create_table",
        sql="""
            CREATE TABLE IF NOT EXISTS my_odbc_table (
                id INT IDENTITY(1,1) PRIMARY KEY,
                value VARCHAR(255)
            );
        """,
        conn_id="my_odbc_conn",
    )

    insert_data = SQLExecuteQueryOperator(
        task_id="insert_data",
        sql="""
            INSERT INTO my_odbc_table (value) VALUES ('test_value_{{ ds }}');
        """,
        conn_id="my_odbc_conn",
    )

    # Note: For fetching data, you'd typically use OdbcHook in a PythonOperator
    # or a custom operator, as SQLExecuteQueryOperator is for execution.

    create_table >> insert_data