[EH] EH Pylint-Next Cleanup (#35347)

* remove unused import

* removed typing import from TYPE_CHECKING

* add return type and returns to docstrings

* remove unnecessary parens

* add timeout to flush methods in producer

* disable inconsistent return rule

* add timeout kwarg to public api

* add update_token to create_token

* update mgmt_request operation kwargs

* add typed kwargs for send*

* typed kwargs for create_batch

* update receive batch kwargs

* kwargs for mgmt exec op

* typed kwargs for receieve

* add typed kwargs for sync consumer client

* consumer from conn str kwargs

* producer conn str kwargs

* kwargs for send client in transports

* receive client kwargs

* typed kwargs for create_connection

* fix error in event processor

* make locales optional

* fix type in field name

* make max_batch_size position

* pass in timeout kwarg

* address PR comments

* missing comma

* fix uamqp transport

* add final new line

* fix missing comma

* initial mypy fixes

* match up conn manager with amqp.create_connection

* intial mypy fixes for transports

* futher mypy match up conn manager and transport

* fix eventprocessor mypy errors

* further fixes to align typing

* more mypy fixes

* address comments

* add forward ref types for uamqp and azure amqp

* enable uamqp tests for ci run

* disable uAMQP tests

Author: kashifkhan

sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py

@@ -401,7 +401,7 @@ def _backoff(
             )
             raise last_exception
 
-    def _management_request(
+    def _management_request(# pylint:disable=inconsistent-return-statements
         self, mgmt_msg: Union[uamqp_Message, Message], op_type: bytes
     ) -> Any:
         # pylint:disable=assignment-from-none
@@ -414,7 +414,7 @@ def _management_request(
             )
             try:
                 conn = self._conn_manager.get_connection(  # pylint:disable=assignment-from-none
-                    host=self._address.hostname, auth=mgmt_auth
+                    endpoint=self._address.hostname, auth=mgmt_auth
                 )
                 mgmt_client.open(connection=conn)
                 while not mgmt_client.client_ready():
@@ -556,7 +556,7 @@ def _open(self) -> bool:
             auth = self._client._create_auth()
             self._create_handler(auth)
             conn = self._client._conn_manager.get_connection(  # pylint: disable=protected-access
-                host=self._client._address.hostname, auth=auth
+                endpoint=self._client._address.hostname, auth=auth
             )
             self._handler.open(connection=conn)
             while not self._handler.client_ready():
@@ -584,7 +584,7 @@ def _handle_exception(self, exception, *, is_consumer=False):
             exception, self, is_consumer=is_consumer
         )
 
-    def _do_retryable_operation(self, operation, timeout=None, **kwargs):
+    def _do_retryable_operation(self, operation, timeout=None, **kwargs): # pylint:disable=inconsistent-return-statements
         # pylint:disable=protected-access
         timeout_time = (time.time() + timeout) if timeout else None
         retried_times = 0

sdk/eventhub/azure-eventhub/azure/eventhub/_connection_manager.py

@@ -4,7 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 from __future__ import annotations
-from typing import TYPE_CHECKING, Optional, Union, Any
+from typing import TYPE_CHECKING, Dict, Optional, Union, Any
 from threading import Lock
 from enum import Enum
 
@@ -30,9 +30,8 @@ class ConnectionManager(Protocol):
         def get_connection(
             self,
             *,
-            host: Optional[str] = None,
+            endpoint: str,
             auth: Optional[Union[JWTTokenAuth, uamqp_JWTTokenAuth]] = None,
-            endpoint: Optional[str] = None,
         ) -> Union[Connection, uamqp_Connection]:
             pass
 
@@ -49,37 +48,52 @@ class _ConnectionMode(Enum):
 
 
 class _SharedConnectionManager:  # pylint:disable=too-many-instance-attributes
-    def __init__(self, **kwargs: Any):
+    def __init__( #pylint:disable=unused-argument
+            self,
+            *,
+            container_id: Optional[str] = None,
+            custom_endpoint_address: Optional[str] = None,
+            debug: bool = False,
+            error_policy: Optional[Any] = None,
+            properties: Optional[Dict[str, Any]] = None,
+            encoding: str = "UTF-8",
+            transport_type: TransportType = TransportType.Amqp,
+            http_proxy: Optional[str] = None,
+            max_frame_size: int,
+            channel_max: int,
+            idle_timeout: float,
+            remote_idle_timeout_empty_frame_send_ratio: float,
+            amqp_transport: AmqpTransport,
+            **kwargs: Any
+            ):
         self._lock = Lock()
         self._conn: Union[Connection, uamqp_Connection] = None
 
-        self._container_id = kwargs.get("container_id")
-        self._custom_endpoint_address = kwargs.get("custom_endpoint_address")
-        self._debug = kwargs.get("debug")
-        self._error_policy = kwargs.get("error_policy")
-        self._properties = kwargs.get("properties")
-        self._encoding = kwargs.get("encoding") or "UTF-8"
-        self._transport_type = kwargs.get("transport_type") or TransportType.Amqp
-        self._http_proxy = kwargs.get("http_proxy")
-        self._max_frame_size = kwargs.get("max_frame_size")
-        self._channel_max = kwargs.get("channel_max")
-        self._idle_timeout = kwargs.get("idle_timeout")
-        self._remote_idle_timeout_empty_frame_send_ratio = kwargs.get("remote_idle_timeout_empty_frame_send_ratio")
-        self._amqp_transport: AmqpTransport = kwargs.pop("amqp_transport")
+        self._container_id = container_id
+        self._custom_endpoint_address = custom_endpoint_address
+        self._debug = debug
+        self._error_policy = error_policy
+        self._properties = properties
+        self._encoding = encoding
+        self._transport_type = transport_type
+        self._http_proxy = http_proxy
+        self._max_frame_size = max_frame_size
+        self._channel_max = channel_max
+        self._idle_timeout = idle_timeout
+        self._remote_idle_timeout_empty_frame_send_ratio = remote_idle_timeout_empty_frame_send_ratio
+        self._amqp_transport: AmqpTransport = amqp_transport
 
     def get_connection(
         self,
         *,
-        host: Optional[str] = None,
+        endpoint: str,
         auth: Optional[Union[JWTTokenAuth, uamqp_JWTTokenAuth]] = None,
-        endpoint: Optional[str] = None,
     ) -> Union[Connection, uamqp_Connection]:
         with self._lock:
             if self._conn is None:
                 self._conn = self._amqp_transport.create_connection(
-                    host=host,
-                    auth=auth,
                     endpoint=endpoint,
+                    auth=auth,
                     custom_endpoint_address=self._custom_endpoint_address,
                     container_id=self._container_id,
                     max_frame_size=self._max_frame_size,
@@ -113,9 +127,8 @@ def __init__(self, **kwargs):
     def get_connection(  # pylint:disable=unused-argument
         self,
         *,
-        host: Optional[str] = None,
+        endpoint: str,
         auth: Optional[Union[JWTTokenAuth, uamqp_JWTTokenAuth]] = None,
-        endpoint: Optional[str] = None,
     ) -> None:
         return None
 

sdk/eventhub/azure-eventhub/azure/eventhub/_consumer.py

@@ -159,7 +159,7 @@ def _create_handler(self, auth: Union[uamqp_JWTTokenAuth, JWTTokenAuth]) -> None
             auth=auth,
             network_trace=self._client._config.network_tracing,  # pylint:disable=protected-access
             link_credit=self._prefetch,
-            link_properties=self._link_properties,
+            link_properties=self._link_properties, # type: ignore
             timeout=self._timeout,
             idle_timeout=self._idle_timeout,
             retry_policy=self._retry_policy,
@@ -203,7 +203,7 @@ def _open(self) -> bool:
             auth = self._client._create_auth()
             self._create_handler(auth)
             conn = self._client._conn_manager.get_connection(  # pylint: disable=protected-access
-                host=self._client._address.hostname, auth=auth
+                endpoint=self._client._address.hostname, auth=auth
             )
             self._handler = cast("ReceiveClient", self._handler)
             self._handler.open(connection=conn)