Add support to create connections using uri in SDK

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -896,6 +896,7 @@ async def test_trigger_can_call_variables_connections_and_xcoms_methods(session,
                 "password": "pass",
                 "port": 443,
                 "extra": '{"key": "value"}',
+                "uri": None,
             },
             "variable": {
                 "get_variable": "some_variable_value",

task-sdk/src/airflow/sdk/definitions/connection.py

@@ -108,6 +108,7 @@ class Connection:
     :param port: The port number.
     :param extra: Extra metadata. Non-standard data such as private/SSH keys can be saved here. JSON
         encoded object.
+    :param uri: URI address describing connection parameters.
     """
 
     conn_id: str
@@ -119,9 +120,26 @@ class Connection:
     password: str | None = None
     port: int | None = None
     extra: str | None = None
+    uri: str | None = None
 
     EXTRA_KEY = "__extra__"
 
+    _URI_FORBIDDEN_FIELDS = ("conn_type", "host", "login", "password", "schema", "port", "extra")
+
+    def __attrs_post_init__(self) -> None:
+        if self.uri is not None:
+            if any(getattr(self, f) for f in self._URI_FORBIDDEN_FIELDS):
+                raise AirflowException(
+                    "You must create an object using the URI or individual values "
+                    "(conn_type, host, login, password, schema, port or extra). "
+                    "You can't mix these two ways to create this object."
+                )
+            conn_from_uri = self.from_uri(self.uri, conn_id=self.conn_id)
+            for attr in attrs.fields(type(conn_from_uri)):
+                if attr.name != "uri":
+                    object.__setattr__(self, attr.name, getattr(conn_from_uri, attr.name))
+            object.__setattr__(self, "uri", None)
+
     def get_uri(self) -> str:
         """Generate and return connection in URI format."""
         from urllib.parse import parse_qsl
@@ -265,6 +283,8 @@ def to_dict(self, *, prune_empty: bool = False, validate: bool = True) -> dict[s
         """
         Convert Connection to json-serializable dictionary.
 
+        Note: uri is init-only and is excluded from serialization.
+
         :param prune_empty: Whether or not remove empty values.
         :param validate: Validate dictionary is JSON-serializable
 
@@ -292,6 +312,7 @@ def to_dict(self, *, prune_empty: bool = False, validate: bool = True) -> dict[s
     @classmethod
     def from_json(cls, value, conn_id=None) -> Connection:
         kwargs = json.loads(value)
+        kwargs.pop("uri", None)  # uri is init-only, never deserialize
         extra = kwargs.pop("extra", None)
         if extra:
             kwargs["extra"] = extra if isinstance(extra, str) else json.dumps(extra)

task-sdk/tests/task_sdk/definitions/test_connection.py

@@ -377,6 +377,19 @@ def test_from_uri_invalid_protocol_host_error(self):
         with pytest.raises(AirflowException, match="Invalid connection string"):
             Connection.from_uri(uri, conn_id="test_conn")
 
+    def test_connection_constructor_with_uri(self):
+        """Test Connection(uri=..., conn_id=...) constructor form."""
+        conn = Connection(conn_id="test_conn", uri="postgres://user:pass@host:5432/db")
+
+        assert conn.conn_id == "test_conn"
+        assert conn.conn_type == "postgres"
+        assert conn.host == "host"
+        assert conn.login == "user"
+        assert conn.password == "pass"
+        assert conn.port == 5432
+        assert conn.schema == "db"
+        assert conn.uri is None
+
     def test_from_uri_roundtrip(self):
         """Test that from_uri and get_uri are inverse operations."""
         original_uri = "postgres://user:pass@host:5432/db?param1=value1&param2=value2"