From 8e12918bbd77842ced2697a7ab8cd82f03ee9e59 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 18:45:30 +0545
Subject: [PATCH 01/13] Add HttpHookAsync implementation

---
 airflow/providers/http/hooks/http.py    | 148 +++++++++++++++++++++++-
 airflow/providers/http/provider.yaml    |   2 +
 generated/provider_dependencies.json    |   2 +
 tests/providers/http/hooks/test_http.py |  83 ++++++++++++-
 4 files changed, 233 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 6c37818ccb155..3bceca72e3987 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -17,16 +17,23 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any, Callable
+import asyncio
+from typing import TYPE_CHECKING, Any, Callable
 
+import aiohttp
 import requests
 import tenacity
+from aiohttp import ClientResponseError
+from asgiref.sync import sync_to_async
 from requests.auth import HTTPBasicAuth
 from requests_toolbelt.adapters.socket_options import TCPKeepAliveAdapter
 
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
 
+if TYPE_CHECKING:
+    from aiohttp.client_reqrep import ClientResponse
+
 
 class HttpHook(BaseHook):
     """
@@ -246,3 +253,142 @@ def test_connection(self):
             return True, "Connection successfully tested"
         except Exception as e:
             return False, str(e)
+
+
+class HttpHookAsync(BaseHook):
+    """
+    Interact with HTTP servers using Python Async.
+
+    :param method: the API method to be called
+    :param http_conn_id: http connection id that has the base
+        API url i.e https://www.google.com/ and optional authentication credentials. Default
+        headers can also be specified in the Extra field in json format.
+    :param auth_type: The auth type for the service
+    """
+
+    conn_name_attr = "http_conn_id"
+    default_conn_name = "http_default"
+    conn_type = "http"
+    hook_name = "HTTP"
+
+    def __init__(
+        self,
+        method: str = "POST",
+        http_conn_id: str = default_conn_name,
+        auth_type: Any = aiohttp.BasicAuth,
+        retry_limit: int = 3,
+        retry_delay: float = 1.0,
+    ) -> None:
+        self.http_conn_id = http_conn_id
+        self.method = method.upper()
+        self.base_url: str = ""
+        self._retry_obj: Callable[..., Any]
+        self.auth_type: Any = auth_type
+        if retry_limit < 1:
+            raise ValueError("Retry limit must be greater than equal to 1")
+        self.retry_limit = retry_limit
+        self.retry_delay = retry_delay
+
+    async def run(
+        self,
+        endpoint: str | None = None,
+        data: dict[str, Any] | str | None = None,
+        headers: dict[str, Any] | None = None,
+        extra_options: dict[str, Any] | None = None,
+    ) -> "ClientResponse":
+        r"""
+        Performs 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)
+
+            if conn.host and "://" in conn.host:
+                self.base_url = conn.host
+            else:
+                # schema defaults to HTTP
+                schema = conn.schema if conn.schema else "http"
+                host = conn.host if conn.host else ""
+                self.base_url = schema + "://" + host
+
+            if conn.port:
+                self.base_url = self.base_url + ":" + str(conn.port)
+            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:
+                raise AirflowException(f"Unexpected HTTP Method: {self.method}")
+
+            attempt_num = 1
+            while True:
+                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 response
+                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 with status: %s", e.status)
+                        # In this case, the user probably made a mistake.
+                        # Don't retry.
+                        raise AirflowException(str(e.status) + ":" + e.message)
+
+                attempt_num += 1
+                await asyncio.sleep(self.retry_delay)
+
+    def _retryable_error_async(self, exception: ClientResponseError) -> bool:
+        """
+        Determines whether or not an exception that was thrown might be successful
+        on a subsequent attempt.
+
+        It considers the following to be retryable:
+            - requests_exceptions.ConnectionError
+            - requests_exceptions.Timeout
+            - anything with a status code >= 500
+
+        Most retryable errors are covered by status code >= 500.
+        """
+        return exception.status >= 500
diff --git a/airflow/providers/http/provider.yaml b/airflow/providers/http/provider.yaml
index 4c9f6c3e77764..3294ac56ec310 100644
--- a/airflow/providers/http/provider.yaml
+++ b/airflow/providers/http/provider.yaml
@@ -42,6 +42,8 @@ dependencies:
   # release it as a requirement for airflow
   - requests>=2.26.0
   - requests_toolbelt
+  - aiohttp
+  - asgiref
 
 integrations:
   - integration-name: Hypertext Transfer Protocol (HTTP)
diff --git a/generated/provider_dependencies.json b/generated/provider_dependencies.json
index 90f878796bb95..6bab4f695b471 100644
--- a/generated/provider_dependencies.json
+++ b/generated/provider_dependencies.json
@@ -410,6 +410,8 @@
   },
   "http": {
     "deps": [
+      "aiohttp",
+      "asgiref",
       "requests>=2.26.0",
       "requests_toolbelt"
     ],
diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index 03bf09075c3c8..c770cf4365250 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 from collections import OrderedDict
 from http import HTTPStatus
@@ -30,7 +31,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.http.hooks.http import HttpHook, HttpHookAsync
 
 
 def get_airflow_connection(unused_conn_id=None):
@@ -391,4 +392,84 @@ def test_keep_alive_disabled(self):
             http_send.assert_called()
 
 
+@pytest.mark.asyncio
+async def test_do_api_call_async_non_retryable_error(aioresponse):
+    """Test api call asynchronously with non retryable error."""
+    hook = HttpHookAsync(method="GET")
+    aioresponse.get("http://httpbin.org/non_existent_endpoint", status=400)
+
+    with pytest.raises(AirflowException) as exc, mock.patch.dict(
+        "os.environ",
+        AIRFLOW_CONN_HTTP_DEFAULT="http://httpbin.org/",
+    ):
+        await hook.run(endpoint="non_existent_endpoint")
+
+    assert str(exc.value) == "400:Bad Request"
+
+
+@pytest.mark.asyncio
+async def test_do_api_call_async_retryable_error(caplog, aioresponse):
+    """Test api call asynchronously with retryable error."""
+    caplog.set_level(logging.WARNING, logger="airflow.providers.http.hooks.http")
+    hook = HttpHookAsync(method="GET")
+    aioresponse.get("http://httpbin.org/non_existent_endpoint", status=500, repeat=True)
+
+    with pytest.raises(AirflowException) as exc, mock.patch.dict(
+        "os.environ",
+        AIRFLOW_CONN_HTTP_DEFAULT="http://httpbin.org/",
+    ):
+        await hook.run(endpoint="non_existent_endpoint")
+
+    assert str(exc.value) == "500:Internal Server Error"
+    assert "[Try 3 of 3] Request to http://httpbin.org/non_existent_endpoint failed" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_do_api_call_async_unknown_method():
+    """Test api call asynchronously for unknown method."""
+    hook = HttpHookAsync(method="NOPE")
+    json = {
+        "existing_cluster_id": "xxxx-xxxxxx-xxxxxx",
+    }
+
+    with pytest.raises(AirflowException) as exc:
+        await hook.run(endpoint="non_existent_endpoint", data=json)
+
+    assert str(exc.value) == "Unexpected HTTP Method: NOPE"
+
+
+@pytest.mark.asyncio
+async def test_async_post_request(aioresponse):
+    """Test api call asynchronously for POST request."""
+    hook = HttpHookAsync()
+
+    aioresponse.post(
+        "http://test:8080/v1/test",
+        status=200,
+        payload='{"status":{"status": 200}}',
+        reason="OK",
+    )
+
+    with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
+        resp = await hook.run("v1/test")
+        assert resp.status == 200
+
+
+@pytest.mark.asyncio
+async def test_async_post_request_with_error_code(aioresponse):
+    """Test api call asynchronously for POST request with error."""
+    hook = HttpHookAsync()
+
+    aioresponse.post(
+        "http://test:8080/v1/test",
+        status=418,
+        payload='{"status":{"status": 418}}',
+        reason="I am teapot",
+    )
+
+    with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
+        with pytest.raises(AirflowException):
+            await hook.run("v1/test")
+
+
 send_email_test = mock.Mock()

From 90cd247cad25a9b292561ecc1c7e38bb0f1750bc Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 20:05:12 +0545
Subject: [PATCH 02/13] Refactor rename from HttpHookAsync to HttpAsyncHook

---
 airflow/providers/http/hooks/http.py    |  2 +-
 tests/providers/http/hooks/test_http.py | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 3bceca72e3987..010ac9e1f6cdc 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -255,7 +255,7 @@ def test_connection(self):
             return False, str(e)
 
 
-class HttpHookAsync(BaseHook):
+class HttpAsyncHook(BaseHook):
     """
     Interact with HTTP servers using Python Async.
 
diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index c770cf4365250..ca2087b37d519 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -31,7 +31,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
-from airflow.providers.http.hooks.http import HttpHook, HttpHookAsync
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
 
 
 def get_airflow_connection(unused_conn_id=None):
@@ -395,7 +395,7 @@ def test_keep_alive_disabled(self):
 @pytest.mark.asyncio
 async def test_do_api_call_async_non_retryable_error(aioresponse):
     """Test api call asynchronously with non retryable error."""
-    hook = HttpHookAsync(method="GET")
+    hook = HttpAsyncHook(method="GET")
     aioresponse.get("http://httpbin.org/non_existent_endpoint", status=400)
 
     with pytest.raises(AirflowException) as exc, mock.patch.dict(
@@ -411,7 +411,7 @@ async def test_do_api_call_async_non_retryable_error(aioresponse):
 async def test_do_api_call_async_retryable_error(caplog, aioresponse):
     """Test api call asynchronously with retryable error."""
     caplog.set_level(logging.WARNING, logger="airflow.providers.http.hooks.http")
-    hook = HttpHookAsync(method="GET")
+    hook = HttpAsyncHook(method="GET")
     aioresponse.get("http://httpbin.org/non_existent_endpoint", status=500, repeat=True)
 
     with pytest.raises(AirflowException) as exc, mock.patch.dict(
@@ -427,7 +427,7 @@ async def test_do_api_call_async_retryable_error(caplog, aioresponse):
 @pytest.mark.asyncio
 async def test_do_api_call_async_unknown_method():
     """Test api call asynchronously for unknown method."""
-    hook = HttpHookAsync(method="NOPE")
+    hook = HttpAsyncHook(method="NOPE")
     json = {
         "existing_cluster_id": "xxxx-xxxxxx-xxxxxx",
     }
@@ -441,7 +441,7 @@ async def test_do_api_call_async_unknown_method():
 @pytest.mark.asyncio
 async def test_async_post_request(aioresponse):
     """Test api call asynchronously for POST request."""
-    hook = HttpHookAsync()
+    hook = HttpAsyncHook()
 
     aioresponse.post(
         "http://test:8080/v1/test",
@@ -458,7 +458,7 @@ async def test_async_post_request(aioresponse):
 @pytest.mark.asyncio
 async def test_async_post_request_with_error_code(aioresponse):
     """Test api call asynchronously for POST request with error."""
-    hook = HttpHookAsync()
+    hook = HttpAsyncHook()
 
     aioresponse.post(
         "http://test:8080/v1/test",

From d28b6d1b06004fc3604781e4e5fcc10ae9b9356b Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 22:52:20 +0545
Subject: [PATCH 03/13] Add aioresponse mock

---
 tests/providers/http/hooks/test_http.py | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index ca2087b37d519..ee8f4ff46b02d 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -27,6 +27,7 @@
 import pytest
 import requests
 import tenacity
+from aioresponses import aioresponses
 from requests.adapters import Response
 
 from airflow.exceptions import AirflowException
@@ -392,6 +393,21 @@ def test_keep_alive_disabled(self):
             http_send.assert_called()
 
 
+send_email_test = mock.Mock()
+
+
+@pytest.fixture
+def aioresponse():
+    """
+    Creates an mock async API response.
+    This comes from a mock library specific to the aiohttp package:
+    https://github.com/pnuckowski/aioresponses
+
+    """
+    with aioresponses() as async_response:
+        yield async_response
+
+
 @pytest.mark.asyncio
 async def test_do_api_call_async_non_retryable_error(aioresponse):
     """Test api call asynchronously with non retryable error."""
@@ -470,6 +486,3 @@ async def test_async_post_request_with_error_code(aioresponse):
     with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
         with pytest.raises(AirflowException):
             await hook.run("v1/test")
-
-
-send_email_test = mock.Mock()

From b550adb63895c9bb03f580a70fd291f886938717 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Wed, 1 Feb 2023 12:03:48 +0545
Subject: [PATCH 04/13] Update airflow/providers/http/hooks/http.py

Co-authored-by: Andrey Anshin <Andrey.Anshin@taragol.is>
---
 airflow/providers/http/hooks/http.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 010ac9e1f6cdc..a32858c0d4b5f 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -374,7 +374,7 @@ async def run(
                         self.log.exception("HTTP error with status: %s", e.status)
                         # In this case, the user probably made a mistake.
                         # Don't retry.
-                        raise AirflowException(str(e.status) + ":" + e.message)
+                        raise AirflowException(f"{e.status}:{e.message}"
 
                 attempt_num += 1
                 await asyncio.sleep(self.retry_delay)

From 921dea71d593d9eac5735c217effe31a1d353a04 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Tue, 7 Feb 2023 01:39:26 +0545
Subject: [PATCH 05/13] Add checks for error status 429 and 413 with
 implementation for HEAD, PUT, DELETE, OPTIONS

---
 airflow/providers/http/hooks/http.py | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index a32858c0d4b5f..235a7f8a2e356 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -347,6 +347,14 @@ async def run(
                 request_func = session.post
             elif self.method == "PATCH":
                 request_func = session.patch
+            elif self.method == "HEAD":
+                request_func = session.head
+            elif self.method == "PUT":
+                request_func = session.put
+            elif self.method == "DELETE":
+                request_func = session.delete
+            elif self.method == "OPTIONS":
+                request_func = session.options
             else:
                 raise AirflowException(f"Unexpected HTTP Method: {self.method}")
 
@@ -374,7 +382,7 @@ async def run(
                         self.log.exception("HTTP error with status: %s", e.status)
                         # In this case, the user probably made a mistake.
                         # Don't retry.
-                        raise AirflowException(f"{e.status}:{e.message}"
+                        raise AirflowException(f"{e.status}:{e.message}")
 
                 attempt_num += 1
                 await asyncio.sleep(self.retry_delay)
@@ -391,4 +399,11 @@ def _retryable_error_async(self, exception: ClientResponseError) -> bool:
 
         Most retryable errors are covered by status code >= 500.
         """
+        if exception.status == 429:
+            # don't retry for too Many Requests
+            return False
+        if exception.status == 413:
+            # don't retry for payload Too Large
+            return False
+
         return exception.status >= 500

From 42b130db43aa0ffb8b91792686e9f5b556676f83 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 18:45:30 +0545
Subject: [PATCH 06/13] Add HttpHookAsync implementation

---
 airflow/providers/http/hooks/http.py    | 148 +++++++++++++++++++++++-
 airflow/providers/http/provider.yaml    |   2 +
 generated/provider_dependencies.json    |   2 +
 tests/providers/http/hooks/test_http.py |  83 ++++++++++++-
 4 files changed, 233 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 6c37818ccb155..3bceca72e3987 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -17,16 +17,23 @@
 # under the License.
 from __future__ import annotations
 
-from typing import Any, Callable
+import asyncio
+from typing import TYPE_CHECKING, Any, Callable
 
+import aiohttp
 import requests
 import tenacity
+from aiohttp import ClientResponseError
+from asgiref.sync import sync_to_async
 from requests.auth import HTTPBasicAuth
 from requests_toolbelt.adapters.socket_options import TCPKeepAliveAdapter
 
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
 
+if TYPE_CHECKING:
+    from aiohttp.client_reqrep import ClientResponse
+
 
 class HttpHook(BaseHook):
     """
@@ -246,3 +253,142 @@ def test_connection(self):
             return True, "Connection successfully tested"
         except Exception as e:
             return False, str(e)
+
+
+class HttpHookAsync(BaseHook):
+    """
+    Interact with HTTP servers using Python Async.
+
+    :param method: the API method to be called
+    :param http_conn_id: http connection id that has the base
+        API url i.e https://www.google.com/ and optional authentication credentials. Default
+        headers can also be specified in the Extra field in json format.
+    :param auth_type: The auth type for the service
+    """
+
+    conn_name_attr = "http_conn_id"
+    default_conn_name = "http_default"
+    conn_type = "http"
+    hook_name = "HTTP"
+
+    def __init__(
+        self,
+        method: str = "POST",
+        http_conn_id: str = default_conn_name,
+        auth_type: Any = aiohttp.BasicAuth,
+        retry_limit: int = 3,
+        retry_delay: float = 1.0,
+    ) -> None:
+        self.http_conn_id = http_conn_id
+        self.method = method.upper()
+        self.base_url: str = ""
+        self._retry_obj: Callable[..., Any]
+        self.auth_type: Any = auth_type
+        if retry_limit < 1:
+            raise ValueError("Retry limit must be greater than equal to 1")
+        self.retry_limit = retry_limit
+        self.retry_delay = retry_delay
+
+    async def run(
+        self,
+        endpoint: str | None = None,
+        data: dict[str, Any] | str | None = None,
+        headers: dict[str, Any] | None = None,
+        extra_options: dict[str, Any] | None = None,
+    ) -> "ClientResponse":
+        r"""
+        Performs 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)
+
+            if conn.host and "://" in conn.host:
+                self.base_url = conn.host
+            else:
+                # schema defaults to HTTP
+                schema = conn.schema if conn.schema else "http"
+                host = conn.host if conn.host else ""
+                self.base_url = schema + "://" + host
+
+            if conn.port:
+                self.base_url = self.base_url + ":" + str(conn.port)
+            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:
+                raise AirflowException(f"Unexpected HTTP Method: {self.method}")
+
+            attempt_num = 1
+            while True:
+                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 response
+                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 with status: %s", e.status)
+                        # In this case, the user probably made a mistake.
+                        # Don't retry.
+                        raise AirflowException(str(e.status) + ":" + e.message)
+
+                attempt_num += 1
+                await asyncio.sleep(self.retry_delay)
+
+    def _retryable_error_async(self, exception: ClientResponseError) -> bool:
+        """
+        Determines whether or not an exception that was thrown might be successful
+        on a subsequent attempt.
+
+        It considers the following to be retryable:
+            - requests_exceptions.ConnectionError
+            - requests_exceptions.Timeout
+            - anything with a status code >= 500
+
+        Most retryable errors are covered by status code >= 500.
+        """
+        return exception.status >= 500
diff --git a/airflow/providers/http/provider.yaml b/airflow/providers/http/provider.yaml
index 4c9f6c3e77764..3294ac56ec310 100644
--- a/airflow/providers/http/provider.yaml
+++ b/airflow/providers/http/provider.yaml
@@ -42,6 +42,8 @@ dependencies:
   # release it as a requirement for airflow
   - requests>=2.26.0
   - requests_toolbelt
+  - aiohttp
+  - asgiref
 
 integrations:
   - integration-name: Hypertext Transfer Protocol (HTTP)
diff --git a/generated/provider_dependencies.json b/generated/provider_dependencies.json
index b3fde3a3a83a3..4c3c62fe9a15b 100644
--- a/generated/provider_dependencies.json
+++ b/generated/provider_dependencies.json
@@ -415,6 +415,8 @@
   },
   "http": {
     "deps": [
+      "aiohttp",
+      "asgiref",
       "requests>=2.26.0",
       "requests_toolbelt"
     ],
diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index 03bf09075c3c8..c770cf4365250 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 from collections import OrderedDict
 from http import HTTPStatus
@@ -30,7 +31,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.http.hooks.http import HttpHook, HttpHookAsync
 
 
 def get_airflow_connection(unused_conn_id=None):
@@ -391,4 +392,84 @@ def test_keep_alive_disabled(self):
             http_send.assert_called()
 
 
+@pytest.mark.asyncio
+async def test_do_api_call_async_non_retryable_error(aioresponse):
+    """Test api call asynchronously with non retryable error."""
+    hook = HttpHookAsync(method="GET")
+    aioresponse.get("http://httpbin.org/non_existent_endpoint", status=400)
+
+    with pytest.raises(AirflowException) as exc, mock.patch.dict(
+        "os.environ",
+        AIRFLOW_CONN_HTTP_DEFAULT="http://httpbin.org/",
+    ):
+        await hook.run(endpoint="non_existent_endpoint")
+
+    assert str(exc.value) == "400:Bad Request"
+
+
+@pytest.mark.asyncio
+async def test_do_api_call_async_retryable_error(caplog, aioresponse):
+    """Test api call asynchronously with retryable error."""
+    caplog.set_level(logging.WARNING, logger="airflow.providers.http.hooks.http")
+    hook = HttpHookAsync(method="GET")
+    aioresponse.get("http://httpbin.org/non_existent_endpoint", status=500, repeat=True)
+
+    with pytest.raises(AirflowException) as exc, mock.patch.dict(
+        "os.environ",
+        AIRFLOW_CONN_HTTP_DEFAULT="http://httpbin.org/",
+    ):
+        await hook.run(endpoint="non_existent_endpoint")
+
+    assert str(exc.value) == "500:Internal Server Error"
+    assert "[Try 3 of 3] Request to http://httpbin.org/non_existent_endpoint failed" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_do_api_call_async_unknown_method():
+    """Test api call asynchronously for unknown method."""
+    hook = HttpHookAsync(method="NOPE")
+    json = {
+        "existing_cluster_id": "xxxx-xxxxxx-xxxxxx",
+    }
+
+    with pytest.raises(AirflowException) as exc:
+        await hook.run(endpoint="non_existent_endpoint", data=json)
+
+    assert str(exc.value) == "Unexpected HTTP Method: NOPE"
+
+
+@pytest.mark.asyncio
+async def test_async_post_request(aioresponse):
+    """Test api call asynchronously for POST request."""
+    hook = HttpHookAsync()
+
+    aioresponse.post(
+        "http://test:8080/v1/test",
+        status=200,
+        payload='{"status":{"status": 200}}',
+        reason="OK",
+    )
+
+    with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
+        resp = await hook.run("v1/test")
+        assert resp.status == 200
+
+
+@pytest.mark.asyncio
+async def test_async_post_request_with_error_code(aioresponse):
+    """Test api call asynchronously for POST request with error."""
+    hook = HttpHookAsync()
+
+    aioresponse.post(
+        "http://test:8080/v1/test",
+        status=418,
+        payload='{"status":{"status": 418}}',
+        reason="I am teapot",
+    )
+
+    with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
+        with pytest.raises(AirflowException):
+            await hook.run("v1/test")
+
+
 send_email_test = mock.Mock()

From 6e13adc548eae245af931e1a6771760d7082ff8b Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 20:05:12 +0545
Subject: [PATCH 07/13] Refactor rename from HttpHookAsync to HttpAsyncHook

---
 airflow/providers/http/hooks/http.py    |  2 +-
 tests/providers/http/hooks/test_http.py | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 3bceca72e3987..010ac9e1f6cdc 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -255,7 +255,7 @@ def test_connection(self):
             return False, str(e)
 
 
-class HttpHookAsync(BaseHook):
+class HttpAsyncHook(BaseHook):
     """
     Interact with HTTP servers using Python Async.
 
diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index c770cf4365250..ca2087b37d519 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -31,7 +31,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
-from airflow.providers.http.hooks.http import HttpHook, HttpHookAsync
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
 
 
 def get_airflow_connection(unused_conn_id=None):
@@ -395,7 +395,7 @@ def test_keep_alive_disabled(self):
 @pytest.mark.asyncio
 async def test_do_api_call_async_non_retryable_error(aioresponse):
     """Test api call asynchronously with non retryable error."""
-    hook = HttpHookAsync(method="GET")
+    hook = HttpAsyncHook(method="GET")
     aioresponse.get("http://httpbin.org/non_existent_endpoint", status=400)
 
     with pytest.raises(AirflowException) as exc, mock.patch.dict(
@@ -411,7 +411,7 @@ async def test_do_api_call_async_non_retryable_error(aioresponse):
 async def test_do_api_call_async_retryable_error(caplog, aioresponse):
     """Test api call asynchronously with retryable error."""
     caplog.set_level(logging.WARNING, logger="airflow.providers.http.hooks.http")
-    hook = HttpHookAsync(method="GET")
+    hook = HttpAsyncHook(method="GET")
     aioresponse.get("http://httpbin.org/non_existent_endpoint", status=500, repeat=True)
 
     with pytest.raises(AirflowException) as exc, mock.patch.dict(
@@ -427,7 +427,7 @@ async def test_do_api_call_async_retryable_error(caplog, aioresponse):
 @pytest.mark.asyncio
 async def test_do_api_call_async_unknown_method():
     """Test api call asynchronously for unknown method."""
-    hook = HttpHookAsync(method="NOPE")
+    hook = HttpAsyncHook(method="NOPE")
     json = {
         "existing_cluster_id": "xxxx-xxxxxx-xxxxxx",
     }
@@ -441,7 +441,7 @@ async def test_do_api_call_async_unknown_method():
 @pytest.mark.asyncio
 async def test_async_post_request(aioresponse):
     """Test api call asynchronously for POST request."""
-    hook = HttpHookAsync()
+    hook = HttpAsyncHook()
 
     aioresponse.post(
         "http://test:8080/v1/test",
@@ -458,7 +458,7 @@ async def test_async_post_request(aioresponse):
 @pytest.mark.asyncio
 async def test_async_post_request_with_error_code(aioresponse):
     """Test api call asynchronously for POST request with error."""
-    hook = HttpHookAsync()
+    hook = HttpAsyncHook()
 
     aioresponse.post(
         "http://test:8080/v1/test",

From 624e189d25b3591167cb419c8492fe89e608a7fc Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Thu, 19 Jan 2023 22:52:20 +0545
Subject: [PATCH 08/13] Add aioresponse mock

---
 tests/providers/http/hooks/test_http.py | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index ca2087b37d519..ee8f4ff46b02d 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -27,6 +27,7 @@
 import pytest
 import requests
 import tenacity
+from aioresponses import aioresponses
 from requests.adapters import Response
 
 from airflow.exceptions import AirflowException
@@ -392,6 +393,21 @@ def test_keep_alive_disabled(self):
             http_send.assert_called()
 
 
+send_email_test = mock.Mock()
+
+
+@pytest.fixture
+def aioresponse():
+    """
+    Creates an mock async API response.
+    This comes from a mock library specific to the aiohttp package:
+    https://github.com/pnuckowski/aioresponses
+
+    """
+    with aioresponses() as async_response:
+        yield async_response
+
+
 @pytest.mark.asyncio
 async def test_do_api_call_async_non_retryable_error(aioresponse):
     """Test api call asynchronously with non retryable error."""
@@ -470,6 +486,3 @@ async def test_async_post_request_with_error_code(aioresponse):
     with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_connection):
         with pytest.raises(AirflowException):
             await hook.run("v1/test")
-
-
-send_email_test = mock.Mock()

From 366f451c73c0876484b90081798ad33958cc6545 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Wed, 1 Feb 2023 12:03:48 +0545
Subject: [PATCH 09/13] Update airflow/providers/http/hooks/http.py

Co-authored-by: Andrey Anshin <Andrey.Anshin@taragol.is>
---
 airflow/providers/http/hooks/http.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index 010ac9e1f6cdc..a32858c0d4b5f 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -374,7 +374,7 @@ async def run(
                         self.log.exception("HTTP error with status: %s", e.status)
                         # In this case, the user probably made a mistake.
                         # Don't retry.
-                        raise AirflowException(str(e.status) + ":" + e.message)
+                        raise AirflowException(f"{e.status}:{e.message}"
 
                 attempt_num += 1
                 await asyncio.sleep(self.retry_delay)

From 44a8ac77510cbeb1743caa0cb16400b7ee1dc390 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Tue, 7 Feb 2023 01:39:26 +0545
Subject: [PATCH 10/13] Add checks for error status 429 and 413 with
 implementation for HEAD, PUT, DELETE, OPTIONS

---
 airflow/providers/http/hooks/http.py | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/airflow/providers/http/hooks/http.py b/airflow/providers/http/hooks/http.py
index a32858c0d4b5f..235a7f8a2e356 100644
--- a/airflow/providers/http/hooks/http.py
+++ b/airflow/providers/http/hooks/http.py
@@ -347,6 +347,14 @@ async def run(
                 request_func = session.post
             elif self.method == "PATCH":
                 request_func = session.patch
+            elif self.method == "HEAD":
+                request_func = session.head
+            elif self.method == "PUT":
+                request_func = session.put
+            elif self.method == "DELETE":
+                request_func = session.delete
+            elif self.method == "OPTIONS":
+                request_func = session.options
             else:
                 raise AirflowException(f"Unexpected HTTP Method: {self.method}")
 
@@ -374,7 +382,7 @@ async def run(
                         self.log.exception("HTTP error with status: %s", e.status)
                         # In this case, the user probably made a mistake.
                         # Don't retry.
-                        raise AirflowException(f"{e.status}:{e.message}"
+                        raise AirflowException(f"{e.status}:{e.message}")
 
                 attempt_num += 1
                 await asyncio.sleep(self.retry_delay)
@@ -391,4 +399,11 @@ def _retryable_error_async(self, exception: ClientResponseError) -> bool:
 
         Most retryable errors are covered by status code >= 500.
         """
+        if exception.status == 429:
+            # don't retry for too Many Requests
+            return False
+        if exception.status == 413:
+            # don't retry for payload Too Large
+            return False
+
         return exception.status >= 500

From 00098761bd3c4dcbe6d34885bdba03594493e65a Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Wed, 8 Feb 2023 19:14:16 +0545
Subject: [PATCH 11/13] Add aioresponse in dependency

---
 airflow/providers/http/provider.yaml | 1 +
 generated/provider_dependencies.json | 1 +
 2 files changed, 2 insertions(+)

diff --git a/airflow/providers/http/provider.yaml b/airflow/providers/http/provider.yaml
index 3294ac56ec310..613ef3f02b739 100644
--- a/airflow/providers/http/provider.yaml
+++ b/airflow/providers/http/provider.yaml
@@ -44,6 +44,7 @@ dependencies:
   - requests_toolbelt
   - aiohttp
   - asgiref
+  - aioresponses
 
 integrations:
   - integration-name: Hypertext Transfer Protocol (HTTP)
diff --git a/generated/provider_dependencies.json b/generated/provider_dependencies.json
index 4c3c62fe9a15b..46b2ee3529cd4 100644
--- a/generated/provider_dependencies.json
+++ b/generated/provider_dependencies.json
@@ -416,6 +416,7 @@
   "http": {
     "deps": [
       "aiohttp",
+      "aioresponses",
       "asgiref",
       "requests>=2.26.0",
       "requests_toolbelt"

From 272ef8f9bfc0f934a5b1abfa0911e6a4d9ab4e58 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Mon, 13 Feb 2023 15:28:20 +0545
Subject: [PATCH 12/13] remove aioresponses from providers.yaml to setup.py

---
 airflow/providers/http/provider.yaml | 1 -
 generated/provider_dependencies.json | 1 -
 setup.py                             | 1 +
 3 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/airflow/providers/http/provider.yaml b/airflow/providers/http/provider.yaml
index 613ef3f02b739..3294ac56ec310 100644
--- a/airflow/providers/http/provider.yaml
+++ b/airflow/providers/http/provider.yaml
@@ -44,7 +44,6 @@ dependencies:
   - requests_toolbelt
   - aiohttp
   - asgiref
-  - aioresponses
 
 integrations:
   - integration-name: Hypertext Transfer Protocol (HTTP)
diff --git a/generated/provider_dependencies.json b/generated/provider_dependencies.json
index 46b2ee3529cd4..4c3c62fe9a15b 100644
--- a/generated/provider_dependencies.json
+++ b/generated/provider_dependencies.json
@@ -416,7 +416,6 @@
   "http": {
     "deps": [
       "aiohttp",
-      "aioresponses",
       "asgiref",
       "requests>=2.26.0",
       "requests_toolbelt"
diff --git a/setup.py b/setup.py
index f9da876aece3b..882d01ff60225 100644
--- a/setup.py
+++ b/setup.py
@@ -404,6 +404,7 @@ def write_version(filename: str = str(AIRFLOW_SOURCES_ROOT / "airflow" / "git_ve
     "twine",
     "wheel",
     "yamllint",
+    "aioresponses",
 ]
 
 

From 0a1e683db65440452c5c8317c41798f12a8823c0 Mon Sep 17 00:00:00 2001
From: Ankit Chaurasia <8670962+sunank200@users.noreply.github.com>
Date: Tue, 14 Feb 2023 17:41:03 +0545
Subject: [PATCH 13/13] Remove the external URL

---
 tests/providers/http/hooks/test_http.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/providers/http/hooks/test_http.py b/tests/providers/http/hooks/test_http.py
index ee8f4ff46b02d..820c45c5a39d8 100644
--- a/tests/providers/http/hooks/test_http.py
+++ b/tests/providers/http/hooks/test_http.py
@@ -428,7 +428,6 @@ async def test_do_api_call_async_retryable_error(caplog, aioresponse):
     """Test api call asynchronously with retryable error."""
     caplog.set_level(logging.WARNING, logger="airflow.providers.http.hooks.http")
     hook = HttpAsyncHook(method="GET")
-    aioresponse.get("http://httpbin.org/non_existent_endpoint", status=500, repeat=True)
 
     with pytest.raises(AirflowException) as exc, mock.patch.dict(
         "os.environ",