support Yandex SDK feature "endpoint" (#29635)

* support endpoint field in airflow
---------

Co-authored-by: sofya generalova <sageneralova@yandex-team.ru>

airflow/providers/yandex/hooks/yandex.py

@@ -78,6 +78,11 @@ def get_connection_form_widgets() -> dict[str, Any]:
                 description="Optional. This key will be placed to all created Compute nodes"
                 "to let you have a root shell there",
             ),
+            "endpoint": StringField(
+                lazy_gettext("API endpoint"),
+                widget=BS3TextFieldWidget(),
+                description="Optional. Specify an API endpoint. Leave blank to use default.",
+            ),
         }
 
     @classmethod
@@ -122,7 +127,8 @@ def __init__(
         self.connection = self.get_connection(self.connection_id)
         self.extras = self.connection.extra_dejson
         credentials = self._get_credentials()
-        self.sdk = yandexcloud.SDK(user_agent=self.provider_user_agent(), **credentials)
+        sdk_config = self._get_endpoint()
+        self.sdk = yandexcloud.SDK(user_agent=self.provider_user_agent(), **sdk_config, **credentials)
         self.default_folder_id = default_folder_id or self._get_field("folder_id", False)
         self.default_public_ssh_key = default_public_ssh_key or self._get_field("public_ssh_key", False)
         self.client = self.sdk.client
@@ -145,6 +151,13 @@ def _get_credentials(self) -> dict[str, Any]:
         else:
             return {"token": oauth_token}
 
+    def _get_endpoint(self) -> dict[str, str]:
+        sdk_config = {}
+        endpoint = self._get_field("endpoint", None)
+        if endpoint:
+            sdk_config["endpoint"] = endpoint
+        return sdk_config
+
     def _get_field(self, field_name: str, default: Any = None) -> Any:
         """Get field from extra, first checking short name, then for backcompat we check for prefixed name."""
         if not hasattr(self, "extras"):

docs/apache-airflow-providers-yandex/connections/yandexcloud.rst

@@ -56,3 +56,7 @@ Folder ID (optional)
 
     If specified, this ID will be used by default during creation of nodes and clusters.
     See https://cloud.yandex.com/docs/resource-manager/operations/folder/get-id for details
+
+Endpoint (optional)
+    Set API endpoint
+    See https://github.com/yandex-cloud/python-sdk for default

tests/providers/yandex/hooks/test_yandex.py

@@ -99,6 +99,48 @@ def test_get_field(self, get_credentials_mock, get_connection_mock):
 
         assert hook._get_field("one") == "value_one"
 
+    @mock.patch("airflow.hooks.base.BaseHook.get_connection")
+    @mock.patch("airflow.providers.yandex.hooks.yandex.YandexCloudBaseHook._get_credentials")
+    def test_get_endpoint_specified(self, get_credentials_mock, get_connection_mock):
+        # Inputs to constructor
+        default_folder_id = "test_id"
+        default_public_ssh_key = "test_key"
+
+        extra_dejson = {"endpoint": "my_endpoint", "something_else": "some_value"}
+        get_connection_mock.return_value = mock.Mock(
+            connection_id="yandexcloud_default", extra_dejson=extra_dejson
+        )
+        get_credentials_mock.return_value = {"token": 122323}
+
+        hook = YandexCloudBaseHook(
+            yandex_conn_id=None,
+            default_folder_id=default_folder_id,
+            default_public_ssh_key=default_public_ssh_key,
+        )
+
+        assert hook._get_endpoint() == {"endpoint": "my_endpoint"}
+
+    @mock.patch("airflow.hooks.base.BaseHook.get_connection")
+    @mock.patch("airflow.providers.yandex.hooks.yandex.YandexCloudBaseHook._get_credentials")
+    def test_get_endpoint_unspecified(self, get_credentials_mock, get_connection_mock):
+        # Inputs to constructor
+        default_folder_id = "test_id"
+        default_public_ssh_key = "test_key"
+
+        extra_dejson = {"something_else": "some_value"}
+        get_connection_mock.return_value = mock.Mock(
+            connection_id="yandexcloud_default", extra_dejson=extra_dejson
+        )
+        get_credentials_mock.return_value = {"token": 122323}
+
+        hook = YandexCloudBaseHook(
+            yandex_conn_id=None,
+            default_folder_id=default_folder_id,
+            default_public_ssh_key=default_public_ssh_key,
+        )
+
+        assert hook._get_endpoint() == {}
+
     @pytest.mark.parametrize(
         "uri",
         [