druid_broker_conn = connect(
    ...
    ssl_verify_cert=conn.extra_dejson.get("ssl_verify_cert", True),
)

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.druid.hooks.druid import DruidDbApiHook
from datetime import datetime
from airflow.exceptions import AirflowException
import logging

def test_druid_ssl_verify(conn_id):
    hook = DruidDbApiHook(druid_broker_conn_id=conn_id)
    try:
        conn = hook.get_conn()
        logging.info('=== conn.ssl_verify_cert ===')
        logging.info(conn.ssl_verify_cert)
        cur = conn.cursor()
        cur.execute("SELECT 1")
        result = cur.fetchall()
        logging.info("Druid query result: %s", result)
    except Exception as e:
        logging.error("Connection failed: %s", str(e))
        raise AirflowException(f"SSL test failed for connection {conn_id}: {str(e)}")
    finally:
        conn.close()

with DAG(
    dag_id="test_druid_ssl_verify",
    start_date=datetime(2025, 7, 1),
    schedule=None,
    catchup=False,
    tags=["druid", "ssl-test"],
) as dag:

    test_conn_ssl_true = PythonOperator(
        task_id="test_ssl_verify_true",
        python_callable=test_druid_ssl_verify,
        op_args=["druid_ssl_true"],
    )

    test_conn_ssl_false = PythonOperator(
        task_id="test_ssl_verify_false",
        python_callable=test_druid_ssl_verify,
        op_args=["druid_ssl_false"],
    )

    test_conn_ssl_true >> test_conn_ssl_false

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.druid.hooks.druid import DruidDbApiHook
from datetime import datetime
import logging

def test_druid_get_pandas_df(conn_id: str):
    hook = DruidDbApiHook(druid_broker_conn_id=conn_id)
    try:
        df = hook.get_pandas_df("SELECT 1")
        logging.info("Druid returned dataframe:\n%s", df)
    except Exception as e:
        logging.error("Druid get_pandas_df failed: %s", e)
        raise

with DAG(
    dag_id="test_druid_get_pandas_df",
    start_date=datetime(2025, 7, 1),
    schedule=None,
    catchup=False,
    tags=["druid", "ssl", "get_pandas_df"],
) as dag:

    run_query = PythonOperator(
        task_id="run_druid_query",
        python_callable=test_druid_get_pandas_df,
        op_args=["druid_ssl_false"],
    )