YAML file supports extra json parameters

airflow/secrets/local_filesystem.py

@@ -101,6 +101,7 @@ def _parse_yaml_file(file_path: str) -> Tuple[Dict[str, List[str]], List[FileSyn
         return {}, [FileSyntaxError(line_no=1, message="The file is empty.")]
     try:
         secrets = yaml.safe_load(content)
+
     except yaml.MarkedYAMLError as e:
         return {}, [FileSyntaxError(line_no=e.problem_mark.line, message=str(e))]
     if not isinstance(secrets, dict):
@@ -180,7 +181,7 @@ def _create_connection(conn_id: str, value: Any):
     if isinstance(value, str):
         return Connection(conn_id=conn_id, uri=value)
     if isinstance(value, dict):
-        connection_parameter_names = get_connection_parameter_names()
+        connection_parameter_names = get_connection_parameter_names() | {"extra_dejson"}
         current_keys = set(value.keys())
         if not current_keys.issubset(connection_parameter_names):
             illegal_keys = current_keys - connection_parameter_names
@@ -189,6 +190,14 @@ def _create_connection(conn_id: str, value: Any):
                 f"The object have illegal keys: {illegal_keys_list}. "
                 f"The dictionary can only contain the following keys: {connection_parameter_names}"
             )
+        if "extra" in value and "extra_dejson" in value:
+            raise AirflowException(
+                "The extra and extra_dejson parameters are mutually exclusive. "
+                "Please provide only one parameter."
+            )
+        if "extra_dejson" in value:
+            value["extra"] = json.dumps(value["extra_dejson"])
+            del value["extra_dejson"]
 
         if "conn_id" in current_keys and conn_id != value["conn_id"]:
             raise AirflowException(

docs/howto/use-alternative-secrets-backend.rst

@@ -117,7 +117,7 @@ The following is a sample JSON file.
     }
 
 The YAML file structure is similar to that of a JSON. The key-value pair of connection ID and the definitions of one or more connections.
-The connection can be defined as a URI (string) or JSON object.
+The connection can be defined as a URI (string) or JSON object. Any extra json parameters can be provided with the key extra_dejson (Keys extra and extra_dejson are mutually exclusive).
 For a guide about defining a connection as a URI, see:: :ref:`generating_connection_uri`.
 For a description of the connection object parameters see :class:`~airflow.models.connection.Connection`.
 The following is a sample YAML file.
@@ -137,6 +137,10 @@ The following is a sample YAML file.
       login: Login
       password: None
       port: 1234
+      extra_dejson:
+        a: b
+        nestedblock_dict:
+          x: y
 
 You can also define connections using a ``.env`` file. Then the key is the connection ID, and
 the value should describe the connection using the URI. If the connection ID is repeated, all values will

tests/secrets/test_local_filesystem.py

@@ -226,9 +226,15 @@ def test_missing_file(self, mock_exists):
                schema: lschema
                login: Login
                password: None
-               port: 1234""",
+               port: 1234
+               extra_dejson:
+                 extra__google_cloud_platform__keyfile_dict:
+                   a: b
+                 extra__google_cloud_platform__keyfile_path: asaa""",
                 {"conn_a": ["mysql://hosta"], "conn_b": ["mysql://hostb", "mysql://hostc"],
-                    "conn_c": ["scheme://Login:None@host:1234/lschema"]}),
+                    "conn_c": [''.join("""scheme://Login:None@host:1234/lschema?
+                        extra__google_cloud_platform__keyfile_dict=%7B%27a%27%3A+%27b%27%7D
+                        &extra__google_cloud_platform__keyfile_path=asaa""".split())]}),
         )
     )
     def test_yaml_file_should_load_connection(self, file_content, expected_connection_uris):
@@ -241,6 +247,66 @@ def test_yaml_file_should_load_connection(self, file_content, expected_connectio
 
             self.assertEqual(expected_connection_uris, connection_uris_by_conn_id)
 
+    @parameterized.expand(
+        (
+            ("""conn_c:
+               conn_type: scheme
+               host: host
+               schema: lschema
+               login: Login
+               password: None
+               port: 1234
+               extra_dejson:
+                 aws_conn_id: bbb
+                 region_name: ccc
+                 """, {"conn_c": [{"aws_conn_id": "bbb", "region_name": "ccc"}]}),
+            ("""conn_d:
+               conn_type: scheme
+               host: host
+               schema: lschema
+               login: Login
+               password: None
+               port: 1234
+               extra_dejson:
+                 extra__google_cloud_platform__keyfile_dict:
+                   a: b
+                 extra__google_cloud_platform__key_path: xxx
+                 """, {"conn_d": [{"extra__google_cloud_platform__keyfile_dict": {"a": "b"},
+                                   "extra__google_cloud_platform__key_path": "xxx"}]}),
+
+        )
+    )
+    def test_yaml_file_should_load_connection_extras(self, file_content, expected_extras):
+        with mock_local_file(file_content):
+            connections_by_conn_id = local_filesystem.load_connections("a.yaml")
+            connection_uris_by_conn_id = {
+                conn_id: [connection.extra_dejson for connection in connections]
+                for conn_id, connections in connections_by_conn_id.items()
+            }
+            self.assertEqual(expected_extras, connection_uris_by_conn_id)
+
+    @parameterized.expand(
+        (
+            ("""conn_c:
+               conn_type: scheme
+               host: host
+               schema: lschema
+               login: Login
+               password: None
+               port: 1234
+               extra:
+                 abc: xyz
+               extra_dejson:
+                 aws_conn_id: bbb
+                 region_name: ccc
+                 """, "The extra and extra_dejson parameters are mutually exclusive."),
+        )
+    )
+    def test_yaml_invalid_extra(self, file_content, expected_message):
+        with mock_local_file(file_content):
+            with self.assertRaisesRegex(AirflowException, re.escape(expected_message)):
+                local_filesystem.load_connections("a.yaml")
+
 
 class TestLocalFileBackend(unittest.TestCase):
     def test_should_read_variable(self):