adding oauth configuration

Author: DanielePalaia

rabbitmq_amqp_python_client/connection.py

@@ -35,6 +35,8 @@
     WinSslConfigurationContext,
 )
 
+from .entities import OAuth2Options
+
 logger = logging.getLogger(__name__)
 
 MT = TypeVar("MT")
@@ -60,6 +62,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
+        oauth2_options: Optional[OAuth2Options] = None,
         recovery_configuration: RecoveryConfiguration = RecoveryConfiguration(),
     ):
         """
@@ -93,6 +96,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 +113,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 +130,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):
+
+        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

@@ -15,6 +15,8 @@
     WinSslConfigurationContext,
 )
 
+from .entities import OAuth2Options
+
 logger = logging.getLogger(__name__)
 
 MT = TypeVar("MT")
@@ -41,6 +43,7 @@ def __init__(
         ssl_context: Union[
             PosixSslConfigurationContext, WinSslConfigurationContext, None
         ] = None,
+        oauth2_options: Optional[OAuth2Options] = None,
         recovery_configuration: RecoveryConfiguration = RecoveryConfiguration(),
     ):
         """
@@ -66,6 +69,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,