Configure HttpHook's auth_type from Connection

airflow/models/connection.py

@@ -432,7 +432,7 @@ def get_extra_dejson(self, nested: bool = False) -> dict:
                 self.log.exception("Failed parsing the json for conn_id %s", self.conn_id)
 
             # Mask sensitive keys from this list
-            mask_secret(extra)
+            mask_secret(obj)
 
         return extra
 

airflow/providers_manager.py

@@ -1011,7 +1011,7 @@ def _import_hook(
         :param package_name: provider package - only needed in case connection_type is missing
         : return
         """
-        from wtforms import BooleanField, IntegerField, PasswordField, StringField
+        from wtforms import BooleanField, IntegerField, PasswordField, SelectField, StringField, TextAreaField
 
         if connection_type is None and hook_class_name is None:
             raise ValueError("Either connection_type or hook_class_name must be set")
@@ -1031,7 +1031,14 @@ def _import_hook(
                 raise ValueError(
                     f"Provider package name is not set when hook_class_name ({hook_class_name}) is used"
                 )
-        allowed_field_classes = [IntegerField, PasswordField, StringField, BooleanField]
+        allowed_field_classes = [
+            IntegerField,
+            PasswordField,
+            StringField,
+            BooleanField,
+            TextAreaField,
+            SelectField,
+        ]
         hook_class: type[BaseHook] | None = _correctness_check(package_name, hook_class_name, provider_info)
         if hook_class is None:
             return None

airflow/www/forms.py

@@ -33,6 +33,7 @@
 from flask_appbuilder.forms import DynamicForm
 from flask_babel import lazy_gettext
 from flask_wtf import FlaskForm
+from markupsafe import Markup
 from wtforms import widgets
 from wtforms.fields import Field, IntegerField, PasswordField, SelectField, StringField, TextAreaField
 from wtforms.validators import InputRequired, Optional
@@ -176,6 +177,29 @@ def populate_obj(self, item):
                 field.populate_obj(item, name)
 
 
+class BS3AccordionTextAreaFieldWidget(BS3TextAreaFieldWidget):
+
+    @staticmethod
+    def _make_collapsable_panel(field: Field, content: Markup) -> str:
+        collapsable_id: str = f"collapsable_{field.id}"
+        return f"""
+        <div class="panel panel-default form-panel">
+            <div class="panel-heading">
+                <h4 class="panel-title">
+                    <a class="accordion-toggle" data-toggle="collapse" aria-expanded="false" aria-controls="{collapsable_id}" href="#{collapsable_id}">Show</a>
+                </h4>
+            </div>
+            <div class="panel-collapse collapse" id="{collapsable_id}" aria-expanded="false">
+                {content.__html__()}
+            </div>
+        </div>
+        """
+
+    def __call__(self, field, **kwargs):
+        text_area = super(BS3TextAreaFieldWidget, self).__call__(field, **kwargs)
+        return self._make_collapsable_panel(field=field, content=text_area)
+
+
 @cache
 def create_connection_form_class() -> type[DynamicForm]:
     """
@@ -223,7 +247,7 @@ def process(self, formdata=None, obj=None, **kwargs):
         login = StringField(lazy_gettext("Login"), widget=BS3TextFieldWidget())
         password = PasswordField(lazy_gettext("Password"), widget=BS3PasswordFieldWidget())
         port = IntegerField(lazy_gettext("Port"), validators=[Optional()], widget=BS3TextFieldWidget())
-        extra = TextAreaField(lazy_gettext("Extra"), widget=BS3TextAreaFieldWidget())
+        extra = TextAreaField(lazy_gettext("Extra"), widget=BS3AccordionTextAreaFieldWidget())
 
     for key, value in providers_manager.connection_form_widgets.items():
         setattr(ConnectionForm, key, value.field)

airflow/www/templates/airflow/conn_create.html

@@ -25,7 +25,6 @@
 <meta name="rest_api_enabled" content="{{ rest_api_enabled }}">
 <meta name="test_url"
   content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_connection_endpoint_test_connection') }}">
-
 {# required for codemirror #}
 <link rel="stylesheet" type="text/css" href="{{ url_for_asset('codemirror.css') }}">
 <link rel="stylesheet" type="text/css" href="{{ url_for_asset('lint.css') }}">

docs/spelling_wordlist.txt

@@ -25,6 +25,7 @@ afterall
 AgentKey
 aio
 aiobotocore
+aiohttp
 AioSession
 aiplatform
 Airbnb

providers/apache/druid/src/airflow/providers/apache/druid/__init__.py

@@ -37,3 +37,17 @@
     raise RuntimeError(
         f"The package `apache-airflow-providers-apache-druid:{__version__}` needs Apache Airflow 2.9.0+"
     )
+
+
+def airflow_dependency_version():
+    import re
+    from os.path import dirname, join
+
+    import yaml
+
+    with open(join(dirname(__file__), "provider.yaml"), encoding="utf-8") as file:
+        for dependency in yaml.safe_load(file)["dependencies"]:
+            if dependency.startswith("apache-airflow"):
+                match = re.search(r">=([\d\.]+)", dependency)
+                if match:
+                    return packaging.version.parse(match.group(1))

providers/apache/livy/src/airflow/providers/apache/livy/hooks/livy.py

@@ -18,24 +18,21 @@
 
 from __future__ import annotations
 
-import asyncio
+
 import json
 import re
+import warnings
 from collections.abc import Sequence
 from enum import Enum
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 import aiohttp
 import requests
-from aiohttp import ClientResponseError
-from asgiref.sync import sync_to_async
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
+from airflow.providers.http.exceptions import HttpErrorException
 from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
 
-if TYPE_CHECKING:
-    from airflow.models import Connection
-
 
 class BatchState(Enum):
     """Batch session states."""
@@ -85,6 +82,14 @@ class LivyHook(HttpHook):
     conn_type = "livy"
     hook_name = "Apache Livy"
 
+    @classmethod
+    def get_connection_form_widgets(cls) -> dict[str, Any]:
+        return super().get_connection_form_widgets()
+
+    @classmethod
+    def get_ui_field_behaviour(cls) -> dict[str, Any]:
+        return super().get_ui_field_behaviour()
+
     def __init__(
         self,
         livy_conn_id: str = default_conn_name,
@@ -93,14 +98,10 @@ def __init__(
         auth_type: Any | None = None,
         endpoint_prefix: str | None = None,
     ) -> None:
-        super().__init__()
-        self.method = "POST"
-        self.http_conn_id = livy_conn_id
+        super().__init__(http_conn_id=livy_conn_id, auth_type=auth_type)
         self.extra_headers = extra_headers or {}
         self.extra_options = extra_options or {}
         self.endpoint_prefix = sanitize_endpoint_prefix(endpoint_prefix)
-        if auth_type:
-            self.auth_type = auth_type
 
     def get_conn(self, headers: dict[str, Any] | None = None) -> Any:
         """
@@ -138,7 +139,7 @@ def run_method(
         if not self.extra_options:
             self.extra_options = {"check_response": False}
 
-        back_method = self.method
+        back_method = self.method  # type: ignore
         self.method = method
         try:
             if retry_args:
@@ -508,11 +509,10 @@ def __init__(
         livy_conn_id: str = default_conn_name,
         extra_options: dict[str, Any] | None = None,
         extra_headers: dict[str, Any] | None = None,
+        auth_type: Any | None = None,
         endpoint_prefix: str | None = None,
     ) -> None:
-        super().__init__()
-        self.method = "POST"
-        self.http_conn_id = livy_conn_id
+        super().__init__(http_conn_id=livy_conn_id, auth_type=auth_type)
         self.extra_headers = extra_headers or {}
         self.extra_options = extra_options or {}
         self.endpoint_prefix = sanitize_endpoint_prefix(endpoint_prefix)
@@ -524,96 +524,27 @@ async def _do_api_call_async(
         headers: dict[str, Any] | None = None,
         extra_options: dict[str, Any] | None = None,
     ) -> Any:
-        """
-        Perform an asynchronous HTTP request call.
-
-        :param endpoint: the endpoint to be called i.e. resource/v1/query?
-        :param data: payload to be uploaded or request parameters
-        :param headers: additional headers to be passed through as a dictionary
-        :param extra_options: Additional kwargs to pass when creating a request.
-            For example, ``run(json=obj)`` is passed as ``aiohttp.ClientSession().get(json=obj)``
-        """
-        extra_options = extra_options or {}
-
-        # headers may be passed through directly or in the "extra" field in the connection
-        # definition
-        _headers = {}
-        auth = None
-
-        if self.http_conn_id:
-            conn = await sync_to_async(self.get_connection)(self.http_conn_id)
-
-            self.base_url = self._generate_base_url(conn)
-            if conn.login:
-                auth = self.auth_type(conn.login, conn.password)
-            if conn.extra:
-                try:
-                    _headers.update(conn.extra_dejson)
-                except TypeError:
-                    self.log.warning("Connection to %s has invalid extra field.", conn.host)
-        if headers:
-            _headers.update(headers)
-
-        if self.base_url and not self.base_url.endswith("/") and endpoint and not endpoint.startswith("/"):
-            url = self.base_url + "/" + endpoint
-        else:
-            url = (self.base_url or "") + (endpoint or "")
-
-        async with aiohttp.ClientSession() as session:
-            if self.method == "GET":
-                request_func = session.get
-            elif self.method == "POST":
-                request_func = session.post
-            elif self.method == "PATCH":
-                request_func = session.patch
-            else:
-                return {"Response": f"Unexpected HTTP Method: {self.method}", "status": "error"}
+        warnings.warn(
+            "The '_do_api_call_async' method is deprecated, use 'run_method' instead",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
 
-            for attempt_num in range(1, 1 + self.retry_limit):
-                response = await request_func(
-                    url,
-                    json=data if self.method in ("POST", "PATCH") else None,
-                    params=data if self.method == "GET" else None,
-                    headers=headers,
-                    auth=auth,
-                    **extra_options,
-                )
-                try:
-                    response.raise_for_status()
-                    return await response.json()
-                except ClientResponseError as e:
-                    self.log.warning(
-                        "[Try %d of %d] Request to %s failed.",
-                        attempt_num,
-                        self.retry_limit,
-                        url,
-                    )
-                    if not self._retryable_error_async(e) or attempt_num == self.retry_limit:
-                        self.log.exception("HTTP error, status code: %s", e.status)
-                        # In this case, the user probably made a mistake.
-                        # Don't retry.
-                        return {"Response": {e.message}, "Status Code": {e.status}, "status": "error"}
-
-                await asyncio.sleep(self.retry_delay)
-
-    def _generate_base_url(self, conn: Connection) -> str:
-        if conn.host and "://" in conn.host:
-            base_url: str = conn.host
-        else:
-            # schema defaults to HTTP
-            schema = conn.schema if conn.schema else "http"
-            host = conn.host if conn.host else ""
-            base_url = f"{schema}://{host}"
-        if conn.port:
-            base_url = f"{base_url}:{conn.port}"
-        return base_url
+        return await self.run_method(
+            endpoint=endpoint or "",
+            method=self.method,  # type: ignore
+            data=data,
+            headers=headers,
+            extra_options=extra_options,
+        )
 
     async def run_method(
         self,
         endpoint: str,
         method: str = "GET",
         data: Any | None = None,
         headers: dict[str, Any] | None = None,
+        extra_options: dict[str, Any] | None = None,
     ) -> Any:
         """
         Wrap HttpAsyncHook; allows to change method on the same HttpAsyncHook.
@@ -622,15 +553,26 @@ async def run_method(
         :param endpoint: endpoint
         :param data: request payload
         :param headers: headers
+        :param extra_options: Additional kwargs to pass when creating a request.
         :return: http response
         """
         if method not in ("GET", "POST", "PUT", "DELETE", "HEAD"):
             return {"status": "error", "response": f"Invalid http method {method}"}
 
-        back_method = self.method
+        back_method = self.method  # type: ignore
         self.method = method
         try:
-            result = await self._do_api_call_async(endpoint, data, headers, self.extra_options)
+            async with aiohttp.ClientSession() as session:
+                result = await super().run(
+                    session=session,
+                    endpoint=endpoint,
+                    data=data,
+                    headers=headers,
+                    extra_options=extra_options or self.extra_options,
+                )
+        except HttpErrorException as e:
+            status, message = str(e).split(":", 1)
+            return {"Response": {message}, "Status Code": {status}, "status": "error"}
         finally:
             self.method = back_method
         return {"status": "success", "response": result}