[AIRFLOW-5260] Allow empty uri arguments in connection strings (#5855)

(cherry picked from commit 7fb729d)

airflow/models/connection.py

@@ -143,7 +143,7 @@ def parse_from_uri(self, uri):
             if uri_parts.password else uri_parts.password
         self.port = uri_parts.port
         if uri_parts.query:
-            self.extra = json.dumps(dict(parse_qsl(uri_parts.query)))
+            self.extra = json.dumps(dict(parse_qsl(uri_parts.query, keep_blank_values=True)))
 
     def get_password(self):
         if self._password and self.is_encrypted:

tests/models/test_connection.py

@@ -104,6 +104,19 @@ def test_connection_from_uri_with_extras(self):
         self.assertDictEqual(connection.extra_dejson, {'extra1': 'a value',
                                                        'extra2': '/path/'})
 
+    def test_connection_from_uri_with_empty_extras(self):
+        uri = 'scheme://user:password@host%2flocation:1234/schema?' \
+              'extra1=a%20value&extra2='
+        connection = Connection(uri=uri)
+        self.assertEqual(connection.conn_type, 'scheme')
+        self.assertEqual(connection.host, 'host/location')
+        self.assertEqual(connection.schema, 'schema')
+        self.assertEqual(connection.login, 'user')
+        self.assertEqual(connection.password, 'password')
+        self.assertEqual(connection.port, 1234)
+        self.assertDictEqual(connection.extra_dejson, {'extra1': 'a value',
+                                                       'extra2': ''})
+
     def test_connection_from_uri_with_colon_in_hostname(self):
         uri = 'scheme://user:password@host%2flocation%3ax%3ay:1234/schema?' \
               'extra1=a%20value&extra2=%2fpath%2f'