get_request_headers combomethod (#3467)

* get_request_headers combomethod

* Add newsfragment

Author: kclowes

newsfragments/3467.feature.rst

@@ -0,0 +1 @@
+HTTPProvider and AsyncHTTPProvider's get_request_headers is now available on both the class and the instance

tests/core/providers/test_async_http_provider.py

@@ -94,8 +94,8 @@ async def test_async_user_provided_session() -> None:
     assert cached_session == session
 
 
-def test_get_request_headers():
-    provider = AsyncHTTPProvider()
+@pytest.mark.parametrize("provider", (AsyncHTTPProvider(), AsyncHTTPProvider))
+def test_get_request_headers(provider):
     headers = provider.get_request_headers()
     assert len(headers) == 2
     assert headers["Content-Type"] == "application/json"

tests/core/providers/test_http_provider.py

@@ -101,8 +101,8 @@ def test_user_provided_session():
     assert adapter._pool_maxsize == 20
 
 
-def test_get_request_headers():
-    provider = HTTPProvider()
+@pytest.mark.parametrize("provider", (HTTPProvider(), HTTPProvider))
+def test_get_request_headers(provider):
     headers = provider.get_request_headers()
     assert len(headers) == 2
     assert headers["Content-Type"] == "application/json"

web3/_utils/http.py

@@ -1,9 +1,12 @@
 DEFAULT_HTTP_TIMEOUT = 30.0
 
 
-def construct_user_agent(class_type: type) -> str:
+def construct_user_agent(
+    module: str,
+    class_name: str,
+) -> str:
     from web3 import (
         __version__ as web3_version,
     )
 
-    return f"web3.py/{web3_version}/{class_type.__module__}.{class_type.__qualname__}"
+    return f"web3.py/{web3_version}/{module}.{class_name}"

web3/providers/rpc/async_rpc.py

@@ -19,6 +19,7 @@
     URI,
 )
 from eth_utils import (
+    combomethod,
     to_dict,
 )
 
@@ -108,10 +109,17 @@ def get_request_kwargs(self) -> Iterable[Tuple[str, Any]]:
             yield "headers", self.get_request_headers()
         yield from self._request_kwargs.items()
 
-    def get_request_headers(self) -> Dict[str, str]:
+    @combomethod
+    def get_request_headers(cls) -> Dict[str, str]:
+        if isinstance(cls, AsyncHTTPProvider):
+            cls_name = cls.__class__.__name__
+        else:
+            cls_name = cls.__name__
+        module = cls.__module__
+
         return {
             "Content-Type": "application/json",
-            "User-Agent": construct_user_agent(type(self)),
+            "User-Agent": construct_user_agent(module, cls_name),
         }
 
     async def _make_request(self, method: RPCEndpoint, request_data: bytes) -> bytes:

web3/providers/rpc/rpc.py

@@ -16,6 +16,7 @@
     URI,
 )
 from eth_utils import (
+    combomethod,
     to_dict,
 )
 import requests
@@ -116,10 +117,18 @@ def get_request_kwargs(self) -> Iterable[Tuple[str, Any]]:
             yield "headers", self.get_request_headers()
         yield from self._request_kwargs.items()
 
-    def get_request_headers(self) -> Dict[str, str]:
+    @combomethod
+    def get_request_headers(cls) -> Dict[str, str]:
+        if isinstance(cls, HTTPProvider):
+            cls_name = cls.__class__.__name__
+        else:
+            cls_name = cls.__name__
+
+        module = cls.__module__
+
         return {
             "Content-Type": "application/json",
-            "User-Agent": construct_user_agent(type(self)),
+            "User-Agent": construct_user_agent(module, cls_name),
         }
 
     def _make_request(self, method: RPCEndpoint, request_data: bytes) -> bytes: