adding oauth configuration

Author: DanielePalaia

rabbitmq_amqp_python_client/connection.py

@@ -14,7 +14,11 @@
 
 from .address_helper import validate_address
 from .consumer import Consumer
-from .entities import RecoveryConfiguration, StreamOptions
+from .entities import (
+    OAuth2Options,
+    RecoveryConfiguration,
+    StreamOptions,
+)
 from .exceptions import (
     ArgumentOutOfRangeException,
     ValidationCodeException,
@@ -60,6 +64,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
+        oauth2_options: Optional[OAuth2Options] = None,
         recovery_configuration: RecoveryConfiguration = RecoveryConfiguration(),
     ):
         """
@@ -93,6 +98,7 @@ def __init__(
         self._index: int = -1
         self._publishers: list[Publisher] = []
         self._consumers: list[Consumer] = []
+        self._oauth2_options = oauth2_options
 
         # Some recovery_configuration validation
         if recovery_configuration.back_off_reconnect_interval < timedelta(seconds=1):
@@ -109,19 +115,8 @@ def _set_environment_connection_list(self, connections: []):  # type: ignore
     def _open_connections(self, reconnect_handlers: bool = False) -> None:
 
         logger.debug("inside connection._open_connections creating connection")
-        if self._recovery_configuration.active_recovery is False:
-            self._conn = BlockingConnection(
-                url=self._addr,
-                urls=self._addrs,
-                ssl_domain=self._ssl_domain,
-            )
-        else:
-            self._conn = BlockingConnection(
-                url=self._addr,
-                urls=self._addrs,
-                ssl_domain=self._ssl_domain,
-                on_disconnection_handler=self._on_disconnection,
-            )
+
+        self._create_connection()
 
         if reconnect_handlers is True:
             logger.debug("reconnecting managements, publishers and consumers handlers")
@@ -137,6 +132,35 @@ def _open_connections(self, reconnect_handlers: bool = False) -> None:
                 # Update the broken connection and sender in the consumer
                 self._consumers[i]._update_connection(self._conn)
 
+    def _create_connection(self) -> None:
+
+        user = None
+        password = None
+
+        if self._oauth2_options is not None:
+            user = ""
+            password = self._oauth2_options.token
+
+        if self._recovery_configuration.active_recovery is False:
+            self._conn = BlockingConnection(
+                url=self._addr,
+                urls=self._addrs,
+                oauth2_options=self._oauth2_options,
+                ssl_domain=self._ssl_domain,
+                user=user,
+                password=password,
+            )
+        else:
+            self._conn = BlockingConnection(
+                url=self._addr,
+                urls=self._addrs,
+                oauth2_options=self._oauth2_options,
+                ssl_domain=self._ssl_domain,
+                on_disconnection_handler=self._on_disconnection,
+                user=user,
+                password=password,
+            )
+
     def dial(self) -> None:
         """
         Establish a connection to the AMQP server.

rabbitmq_amqp_python_client/entities.py

@@ -257,3 +257,8 @@ class RecoveryConfiguration:
     active_recovery: bool = True
     back_off_reconnect_interval: timedelta = timedelta(seconds=5)
     MaxReconnectAttempts: int = 5
+
+
+@dataclass
+class OAuth2Options:
+    token: str

rabbitmq_amqp_python_client/environment.py

@@ -9,7 +9,7 @@
 )
 
 from .connection import Connection
-from .entities import RecoveryConfiguration
+from .entities import OAuth2Options, RecoveryConfiguration
 from .ssl_configuration import (
     PosixSslConfigurationContext,
     WinSslConfigurationContext,
@@ -41,6 +41,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
+        oauth2_options: Optional[OAuth2Options] = None,
         recovery_configuration: RecoveryConfiguration = RecoveryConfiguration(),
     ):
         """
@@ -66,6 +67,7 @@ def __init__(
         self._ssl_context = ssl_context
         self._recovery_configuration = recovery_configuration
         self._connections: list[Connection] = []
+        self._oauth2_options = oauth2_options
 
     def connection(
         self,