Enable AWS Secrets Manager backend to retrieve conns using different …

…fields (#18764)

airflow/providers/amazon/aws/secrets/secrets_manager.py

@@ -18,6 +18,7 @@
 """Objects relating to sourcing secrets from AWS Secrets Manager"""
 
 import ast
+import json
 from typing import Optional
 from urllib.parse import urlencode
 
@@ -71,25 +72,31 @@ class SecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
             "conn_type": ["conn_type", "conn_id", "connection_type", "engine"],
         }
 
-    However, these lists can be extended using the configuration parameter ``extra_conn_words``.
+    However, these lists can be extended using the configuration parameter ``extra_conn_words``. Also,
+    you can have a field named extra for extra parameters for the conn. Please note that this extra field
+    must be a valid JSON.
 
     :param connections_prefix: Specifies the prefix of the secret to read to get Connections.
-        If set to None (null), requests for connections will not be sent to AWS Secrets Manager
+        If set to None (null value in the configuration), requests for connections will not be
+        sent to AWS Secrets Manager. If you don't want a connections_prefix, set it as an empty string
     :type connections_prefix: str
     :param variables_prefix: Specifies the prefix of the secret to read to get Variables.
-        If set to None (null), requests for variables will not be sent to AWS Secrets Manager
+        If set to None (null value in the configuration), requests for variables will not be sent to
+        AWS Secrets Manager. If you don't want a variables_prefix, set it as an empty string
     :type variables_prefix: str
     :param config_prefix: Specifies the prefix of the secret to read to get Configurations.
-        If set to None (null), requests for configurations will not be sent to AWS Secrets Manager
+        If set to None (null value in the configuration), requests for configurations will not be sent to
+        AWS Secrets Manager. If you don't want a config_prefix, set it as an empty string
     :type config_prefix: str
     :param profile_name: The name of a profile to use. If not given, then the default profile is used.
     :type profile_name: str
     :param sep: separator used to concatenate secret_prefix and secret_id. Default: "/"
     :type sep: str
     :param full_url_mode: if True, the secrets must be stored as one conn URI in just one field per secret.
-        Otherwise, you can store the secret using different fields (password, user...)
+        If False (set it as false in backend_kwargs), you can store the secret using different
+        fields (password, user...).
     :type full_url_mode: bool
-    :param extra_conn_words: for using just when you set full_url_mode as False and store
+    :param extra_conn_words: for using just when you set full_url_mode as false and store
         the secrets in different fields of secrets manager. You can add more words for each connection
         part beyond the default ones. The extra words to be searched should be passed as a dict of lists,
         each list corresponding to a connection part. The optional keys of the dict must be: user,
@@ -109,15 +116,15 @@ def __init__(
         **kwargs,
     ):
         super().__init__()
-        if connections_prefix is not None:
+        if connections_prefix:
             self.connections_prefix = connections_prefix.rstrip(sep)
         else:
             self.connections_prefix = connections_prefix
-        if variables_prefix is not None:
+        if variables_prefix:
             self.variables_prefix = variables_prefix.rstrip(sep)
         else:
             self.variables_prefix = variables_prefix
-        if config_prefix is not None:
+        if config_prefix:
             self.config_prefix = config_prefix.rstrip(sep)
         else:
             self.config_prefix = config_prefix
@@ -134,12 +141,15 @@ def client(self):
 
         return session.client(service_name="secretsmanager", **self.kwargs)
 
-    def _format_uri_with_extra(self, secret, conn_string):
+    @staticmethod
+    def _format_uri_with_extra(secret, conn_string):
         try:
             extra_dict = secret['extra']
         except KeyError:
             return conn_string
-        conn_string = f"{conn_string}?{urlencode(extra_dict)}"
+
+        extra = json.loads(extra_dict)  # this is needed because extra_dict is a string and we need a dict
+        conn_string = f"{conn_string}?{urlencode(extra)}"
 
         return conn_string
 
@@ -176,10 +186,10 @@ def get_conn_uri(self, conn_id: str):
         :param conn_id: connection id
         :type conn_id: str
         """
-        if self.full_url_mode:
-            if self.connections_prefix is None:
-                return None
+        if self.connections_prefix is None:
+            return None
 
+        if self.full_url_mode:
             return self._get_secret(self.connections_prefix, conn_id)
         else:
             try:
@@ -225,7 +235,10 @@ def _get_secret(self, path_prefix, secret_id: str) -> Optional[str]:
         :param secret_id: Secret Key
         :type secret_id: str
         """
-        secrets_path = self.build_path(path_prefix, secret_id, self.sep)
+        if path_prefix:
+            secrets_path = self.build_path(path_prefix, secret_id, self.sep)
+        else:
+            secrets_path = secret_id
 
         try:
             response = self.client.get_secret_value(

docs/apache-airflow-providers-amazon/secrets-backends/aws-secrets-manager.rst

@@ -19,15 +19,16 @@ AWS Secrets Manager Backend
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 To enable Secrets Manager, specify :py:class:`~airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend`
-as the ``backend`` in  ``[secrets]`` section of ``airflow.cfg``.
+as the ``backend`` in  ``[secrets]`` section of ``airflow.cfg``. These ``backend_kwargs`` are parsed as JSON, hence Python
+values like the bool False or None will be ignored, taking for those kwargs the default values of the secrets backend.
 
 Here is a sample configuration:
 
 .. code-block:: ini
 
     [secrets]
     backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
-    backend_kwargs = {"connections_prefix": "airflow/connections", "variables_prefix": "airflow/variables", "profile_name": "default", "full_url_mode": False}
+    backend_kwargs = {"connections_prefix": "airflow/connections", "variables_prefix": "airflow/variables", "profile_name": "default", "full_url_mode": false}
 
 To authenticate you can either supply a profile name to reference aws profile, e.g. defined in ``~/.aws/config`` or set
 environment variables like ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``.
@@ -36,7 +37,7 @@ environment variables like ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``.
 Storing and Retrieving Connections
 """"""""""""""""""""""""""""""""""
 You can store the different values for a secret in two forms: storing the conn URI in one field (default mode) or using different
-fields in Amazon Secrets Manager (setting ``full_url_mode`` as False in the backend config), as follow:
+fields in Amazon Secrets Manager (setting ``full_url_mode`` as ``false`` in the backend config), as follow:
 .. image:: img/aws-secrets-manager.png
 
 By default you must use some of the following words for each kind of field:
@@ -47,7 +48,8 @@ By default you must use some of the following words for each kind of field:
 * Port: port
 * You should also specify the type of connection, which can be done naming the key as conn_type, conn_id,
   connection_type or engine. Valid values for this field are postgres, mysql, snowflake, google_cloud, mongo...
-* For the extra value of the connections, you have to type a dictionary.
+* For the extra value of the connections, a field called extra must exists. Please note this extra field
+  should be a valid JSON.
 
 However, more words can be added to the list using the parameter ``extra_conn_words`` in the configuration. This
 parameter has to be a dict of lists with the following optional keys: user, password, host, schema, conn_type
@@ -78,6 +80,7 @@ Verify that you can get the secret:
         "CreatedDate": "2020-04-08T02:10:35.132000+01:00"
     }
 
+If you don't want to use any ``connections_prefix`` for retrieving connections, set it as an empty string ``""`` in the configuration.
 
 Storing and Retrieving Variables
 """"""""""""""""""""""""""""""""

tests/providers/amazon/aws/secrets/test_secrets_manager.py

@@ -93,7 +93,7 @@ def test_get_conn_uri_broken_field_mode_extra_words_added(self):
 
     @mock_secretsmanager
     def test_format_uri_with_extra(self):
-        secret = {'extra': {'key1': 'value1', 'key2': 'value2'}}
+        secret = {'extra': '{"key1": "value1", "key2": "value2"}'}
         conn_string = 'CS'
         secrets_manager_backend = SecretsManagerBackend()