IBOR Airflow Providers

公司内部 Apache Airflow providers 的单 wheel 工程。当前提供 OceanBase MySQL 模式、TiDB 和 openGauss 支持，后续 provider 继续放在 src/airflow/providers/ 下，并在统一的 get_provider_info() 中注册。

开发与构建

uv sync
uv run pytest
uv build

工程固定使用 Python 3.10，并与目标环境的 Airflow 2.7.3、MySQL provider 5.4.0、Postgres provider 5.7.1 保持一致。

如需使用 Airflow 官方 Python 3.10 约束同步本地环境，请执行：

uv pip install \
    --python .venv/bin/python \
    --editable . \
    --group dev \
    --constraints constraints-3.10.txt \
    --exact \
    --strict

同步后运行命令时应增加 --no-sync，避免 uv 根据 uv.lock 自动覆盖当前 约束环境，例如：

uv run --no-sync pytest

OceanBase connection

在 Airflow 中创建以下 connection：

• Connection Id：按 DAG 需要填写，例如 oceanbase
• Connection Type：OceanBase
• Host、Database、Login、Password、Port：填写 OceanBase MySQL 租户的连接信息

Provider 固定使用 mysql-connector-python，无需在 Extra 中设置 client。

Hook

from airflow.providers.oceanbase.hooks.oceanbase import OceanBaseHook

hook = OceanBaseHook(oceanbase_conn_id="oceanbase")
result = hook.get_first("SELECT 1")

通用 SQL Operator

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

test_oceanbase = SQLExecuteQueryOperator(
    task_id="test_oceanbase",
    conn_id="oceanbase",
    sql="SELECT 1",
)

TiDB connection

在 Airflow 中创建以下 connection：

• Connection Id：按 DAG 需要填写，例如 tidb
• Connection Type：TiDB
• Host、Database、Login、Password、Port：填写 TiDB 的连接信息

TiDB 默认端口为 4000。Provider 固定使用 mysql-connector-python，无需在 Extra 中设置 client。

Hook

from airflow.providers.tidb.hooks.tidb import TiDBHook

hook = TiDBHook(tidb_conn_id="tidb")
result = hook.get_first("SELECT 1")

通用 SQL Operator

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

test_tidb = SQLExecuteQueryOperator(
    task_id="test_tidb",
    conn_id="tidb",
    sql="SELECT 1",
)

openGauss connection

在 Airflow 中创建以下 connection：

• Connection Id：按 DAG 需要填写，例如 opengauss
• Connection Type：openGauss
• Host、Database、Login、Password、Port：填写 openGauss 的连接信息

默认端口为 5432，实际部署使用其他端口时请显式填写。Provider 固定使用 psycopg2-binary==2.9.9，目标数据库必须允许标准 psycopg2 兼容的认证方式； 本 provider 不包含 openGauss 专用认证驱动。

Extra 支持 psycopg2 连接参数，例如 SSL 和连接超时：

{"sslmode": "verify-full", "sslrootcert": "/certs/ca.pem", "connect_timeout": 10}

可通过 Extra 的 cursor 选择 dictcursor、realdictcursor 或 namedtuplecursor。不支持 Postgres Hook 的 AWS IAM / Redshift 专用连接功能。

Hook

from airflow.providers.opengauss.hooks.opengauss import OpenGaussHook

hook = OpenGaussHook(opengauss_conn_id="opengauss")
result = hook.get_first("SELECT 1")

Hook 支持通过 database 覆盖 Connection 中的数据库，以及通过 options 传入会话配置，例如 options="-c search_path=public"。 get_uri() 返回 postgresql+psycopg2 SQLAlchemy URI。

通用 SQL Operator

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

test_opengauss = SQLExecuteQueryOperator(
    task_id="test_opengauss",
    conn_id="opengauss",
    sql="SELECT 1",
)

扩展新 provider

1. 在 src/airflow/providers/<name>/ 中新增 Hook 或其他组件。
2. 在 airflow.providers.custom.get_provider_info 中注册 connection type。
3. 增加对应测试并提升工程版本号。

所有 provider 共用当前工程的版本和发布周期。

发布到 Nexus

项目依赖从公司 Nexus 的 pypi-public 索引下载，构建产物发布到 pypi-hosted。发行包名为 ibor-airflow-providers，避免与 Nexus 中已有的 同名公开包 airflow-providers 冲突。

发布前通过环境变量提供 Nexus 凭据，不要将凭据写入仓库：

export UV_PUBLISH_USERNAME="<nexus-username>"
export UV_PUBLISH_PASSWORD="<nexus-password>"

构建、检查并发布：

uv build
uv publish --index nexus

发布后可以通过 Nexus 安装指定版本：

uv pip install \
    --index-url https://nexus-tzgl.szkingdom.com/repository/pypi-public/simple \
    ibor-airflow-providers==0.3.0

从旧发行名升级时，应先卸载旧包，避免两个发行版同时注册相同的 Airflow provider 模块：

uv pip uninstall airflow-providers
uv pip install \
    --index-url https://nexus-tzgl.szkingdom.com/repository/pypi-public/simple \
    ibor-airflow-providers==0.3.0