Optionally use client.files_upload_v2 in Slack Provider (#36757)

* Optionally use `client.files_upload_v2` in Slack Provider

* Revert  default value for method

* Remove redundand assigment

(cherry picked from commit 9758acf)

Author: Taragolis

airflow/providers/slack/hooks/slack.py

@@ -21,10 +21,11 @@
 import warnings
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Sequence
+from typing import TYPE_CHECKING, Any, Sequence, TypedDict
 
 from slack_sdk import WebClient
 from slack_sdk.errors import SlackApiError
+from typing_extensions import NotRequired
 
 from airflow.exceptions import AirflowNotFoundException
 from airflow.hooks.base import BaseHook
@@ -36,6 +37,26 @@
     from slack_sdk.web.slack_response import SlackResponse
 
 
+class FileUploadTypeDef(TypedDict):
+    """
+    Represents the structure of the file upload data.
+
+    :ivar file: Optional. Path to file which need to be sent.
+    :ivar content: Optional. File contents. If omitting this parameter, you must provide a file.
+    :ivar filename: Optional. Displayed filename.
+    :ivar title: Optional. The title of the uploaded file.
+    :ivar alt_txt: Optional. Description of image for screen-reader.
+    :ivar snippet_type: Optional. Syntax type of the snippet being uploaded.
+    """
+
+    file: NotRequired[str | None]
+    content: NotRequired[str | None]
+    filename: NotRequired[str | None]
+    title: NotRequired[str | None]
+    alt_txt: NotRequired[str | None]
+    snippet_type: NotRequired[str | None]
+
+
 class SlackHook(BaseHook):
     """
     Creates a Slack API Connection to be used for calls.
@@ -111,6 +132,9 @@ def __init__(
             extra_client_args["logger"] = self.log
         self.extra_client_args = extra_client_args
 
+        # Use for caching channels result
+        self._channels_mapping: dict[str, str] = {}
+
     @cached_property
     def client(self) -> WebClient:
         """Get the underlying slack_sdk.WebClient (cached)."""
@@ -212,6 +236,128 @@ def send_file(
             channels=channels,
         )
 
+    def send_file_v2(
+        self,
+        *,
+        channel_id: str | None = None,
+        file_uploads: FileUploadTypeDef | list[FileUploadTypeDef],
+        initial_comment: str | None = None,
+    ) -> SlackResponse:
+        """
+        Sends one or more files to a Slack channel using the Slack SDK Client method `files_upload_v2`.
+
+        :param channel_id: The ID of the channel to send the file to.
+            If omitting this parameter, then file will send to workspace.
+        :param file_uploads: The file(s) specification to upload.
+        :param initial_comment: The message text introducing the file in specified ``channel``.
+        """
+        if channel_id and channel_id.startswith("#"):
+            retried_channel_id = self.get_channel_id(channel_id[1:])
+            warnings.warn(
+                "The method `files_upload_v2` in the Slack SDK Client expects a Slack Channel ID, "
+                f"but received a Slack Channel Name. To resolve this, consider replacing {channel_id!r} "
+                f"with the corresponding Channel ID {retried_channel_id!r}.",
+                UserWarning,
+                stacklevel=2,
+            )
+            channel_id = retried_channel_id
+
+        if not isinstance(file_uploads, list):
+            file_uploads = [file_uploads]
+        for file_upload in file_uploads:
+            if not file_upload.get("filename"):
+                # Some of early version of Slack SDK (such as 3.19.0) raise an error if ``filename`` not set.
+                file_upload["filename"] = "Uploaded file"
+
+        return self.client.files_upload_v2(
+            channel=channel_id,
+            # mypy doesn't happy even if TypedDict used instead of dict[str, Any]
+            # see: https://github.com/python/mypy/issues/4976
+            file_uploads=file_uploads,  # type: ignore[arg-type]
+            initial_comment=initial_comment,
+        )
+
+    def send_file_v1_to_v2(
+        self,
+        *,
+        channels: str | Sequence[str] | None = None,
+        file: str | Path | None = None,
+        content: str | None = None,
+        filename: str | None = None,
+        initial_comment: str | None = None,
+        title: str | None = None,
+        filetype: str | None = None,
+    ) -> list[SlackResponse]:
+        """
+        Smooth transition between ``send_file`` and ``send_file_v2`` methods.
+
+        :param channels: Comma-separated list of channel names or IDs where the file will be shared.
+            If omitting this parameter, then file will send to workspace.
+            File would be uploaded for each channel individually.
+        :param file: Path to file which need to be sent.
+        :param content: File contents. If omitting this parameter, you must provide a file.
+        :param filename: Displayed filename.
+        :param initial_comment: The message text introducing the file in specified ``channels``.
+        :param title: Title of the file.
+        :param filetype: A file type identifier.
+        """
+        if not exactly_one(file, content):
+            raise ValueError("Either `file` or `content` must be provided, not both.")
+        if file:
+            file = Path(file)
+            file_uploads: FileUploadTypeDef = {"file": file.__fspath__(), "filename": filename or file.name}
+        else:
+            file_uploads = {"content": content, "filename": filename}
+
+        file_uploads.update({"title": title, "snippet_type": filetype})
+
+        if channels:
+            if isinstance(channels, str):
+                channels = channels.split(",")
+            channels_to_share: list[str | None] = list(map(str.strip, channels))
+        else:
+            channels_to_share = [None]
+
+        responses = []
+        for channel in channels_to_share:
+            responses.append(
+                self.send_file_v2(
+                    channel_id=channel, file_uploads=file_uploads, initial_comment=initial_comment
+                )
+            )
+        return responses
+
+    def get_channel_id(self, channel_name: str) -> str:
+        """
+        Retrieves a Slack channel id by a channel name.
+
+        It continuously iterates over all Slack channels (public and private)
+        until it finds the desired channel name in addition cache results for further usage.
+
+        .. seealso::
+            https://api.slack.com/methods/conversations.list
+
+        :param channel_name: The name of the Slack channel for which ID has to be found.
+        """
+        next_cursor = None
+        while not (channel_id := self._channels_mapping.get(channel_name)):
+            res = self.client.conversations_list(cursor=next_cursor, types="public_channel,private_channel")
+            if TYPE_CHECKING:
+                # Slack SDK response type too broad, this should make mypy happy
+                assert isinstance(res.data, dict)
+
+            for channel_data in res.data.get("channels", []):
+                self._channels_mapping[channel_data["name"]] = channel_data["id"]
+
+            if not (next_cursor := res.data.get("response_metadata", {}).get("next_cursor")):
+                channel_id = self._channels_mapping.get(channel_name)
+                break
+
+        if not channel_id:
+            msg = f"Unable to find slack channel with name: {channel_name!r}"
+            raise LookupError(msg)
+        return channel_id
+
     def test_connection(self):
         """Tests the Slack API connection.
 

airflow/providers/slack/operators/slack.py

@@ -22,20 +22,24 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, Sequence
 
+from typing_extensions import Literal
+
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.models import BaseOperator
 from airflow.providers.slack.hooks.slack import SlackHook
 
 if TYPE_CHECKING:
     from slack_sdk.http_retry import RetryHandler
 
+    from airflow.utils.context import Context
+
 
 class SlackAPIOperator(BaseOperator):
     """Base Slack Operator class.
 
     :param slack_conn_id: :ref:`Slack API Connection <howto/connection:slack>`
         which its password is Slack API token.
-    :param method: The Slack API Method to Call (https://api.slack.com/methods). Optional
+    :param method: The Slack API Method to Call (https://api.slack.com/methods).
     :param api_params: API Method call parameters (https://api.slack.com/methods). Optional
     :param timeout: The maximum number of seconds the client will wait to connect
         and receive a response from Slack. Optional
@@ -56,6 +60,13 @@ def __init__(
         retry_handlers: list[RetryHandler] | None = None,
         **kwargs,
     ) -> None:
+        if not method:
+            warnings.warn(
+                "Define `method` parameter as empty string or None is deprecated. "
+                "In the future it will raise an error on task initialisation.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
         super().__init__(**kwargs)
         self.slack_conn_id = slack_conn_id
         self.method = method
@@ -90,7 +101,10 @@ def construct_api_call_params(self) -> Any:
             "SlackAPIOperator should not be used directly. Chose one of the subclasses instead"
         )
 
-    def execute(self, **kwargs):
+    def execute(self, context: Context):
+        if not self.method:
+            msg = f"Expected non empty `method` attribute in {type(self).__name__!r}, but got {self.method!r}"
+            raise ValueError(msg)
         if not self.api_params:
             self.construct_api_call_params()
         self.hook.call(self.method, json=self.api_params)
@@ -139,14 +153,13 @@ def __init__(
         attachments: list | None = None,
         **kwargs,
     ) -> None:
-        self.method = "chat.postMessage"
+        super().__init__(method="chat.postMessage", **kwargs)
         self.channel = channel
         self.username = username
         self.text = text
         self.icon_url = icon_url
         self.attachments = attachments or []
         self.blocks = blocks or []
-        super().__init__(method=self.method, **kwargs)
 
     def construct_api_call_params(self) -> Any:
         self.api_params = {
@@ -192,7 +205,7 @@ class SlackAPIFileOperator(SlackAPIOperator):
     :param filetype: slack filetype. (templated) See: https://api.slack.com/types/file#file_types
     :param content: file content. (templated)
     :param title: title of file. (templated)
-    :param channel: (deprecated) channel in which to sent file on slack name
+    :param method_version: The version of the method of Slack SDK Client to be used, either "v1" or "v2".
     """
 
     template_fields: Sequence[str] = (
@@ -213,10 +226,10 @@ def __init__(
         filetype: str | None = None,
         content: str | None = None,
         title: str | None = None,
-        channel: str | None = None,
+        method_version: Literal["v1", "v2"] = "v1",
         **kwargs,
     ) -> None:
-        if channel:
+        if channel := kwargs.pop("channel", None):
             warnings.warn(
                 "Argument `channel` is deprecated and will removed in a future releases. "
                 "Please use `channels` instead.",
@@ -227,16 +240,23 @@ def __init__(
                 raise ValueError(f"Cannot set both arguments: channel={channel!r} and channels={channels!r}.")
             channels = channel
 
+        super().__init__(method="files.upload", **kwargs)
         self.channels = channels
         self.initial_comment = initial_comment
         self.filename = filename
         self.filetype = filetype
         self.content = content
         self.title = title
-        super().__init__(method="files.upload", **kwargs)
+        self.method_version = method_version
+
+    @property
+    def _method_resolver(self):
+        if self.method_version == "v1":
+            return self.hook.send_file
+        return self.hook.send_file_v1_to_v2
 
-    def execute(self, **kwargs):
-        self.hook.send_file(
+    def execute(self, context: Context):
+        self._method_resolver(
             channels=self.channels,
             # For historical reason SlackAPIFileOperator use filename as reference to file
             file=self.filename,

airflow/providers/slack/provider.yaml

@@ -58,7 +58,7 @@ versions:
 dependencies:
   - apache-airflow>=2.6.0
   - apache-airflow-providers-common-sql>=1.3.1
-  - slack_sdk>=3.0.0
+  - slack_sdk>=3.19.0
 
 integrations:
   - integration-name: Slack

airflow/providers/slack/transfers/sql_to_slack.py

@@ -17,9 +17,12 @@
 from __future__ import annotations
 
 import warnings
+from functools import cached_property
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING, Any, Mapping, Sequence
 
+from typing_extensions import Literal
+
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.slack.hooks.slack import SlackHook
 from airflow.providers.slack.transfers.base_sql_to_slack import BaseSqlToSlackOperator
@@ -53,6 +56,7 @@ class SqlToSlackApiFileOperator(BaseSqlToSlackOperator):
     :param slack_initial_comment: The message text introducing the file in specified ``slack_channels``.
     :param slack_title: Title of file.
     :param slack_base_url: A string representing the Slack API base URL. Optional
+    :param slack_method_version: The version of the Slack SDK Client method to be used, either "v1" or "v2".
     :param df_kwargs: Keyword arguments forwarded to ``pandas.DataFrame.to_{format}()`` method.
     """
 
@@ -81,6 +85,7 @@ def __init__(
         slack_initial_comment: str | None = None,
         slack_title: str | None = None,
         slack_base_url: str | None = None,
+        slack_method_version: Literal["v1", "v2"] = "v1",
         df_kwargs: dict | None = None,
         **kwargs,
     ):
@@ -93,22 +98,33 @@ def __init__(
         self.slack_initial_comment = slack_initial_comment
         self.slack_title = slack_title
         self.slack_base_url = slack_base_url
+        self.slack_method_version = slack_method_version
         self.df_kwargs = df_kwargs or {}
 
+    @cached_property
+    def slack_hook(self):
+        """Slack API Hook."""
+        return SlackHook(
+            slack_conn_id=self.slack_conn_id,
+            base_url=self.slack_base_url,
+            timeout=self.slack_timeout,
+            proxy=self.slack_proxy,
+            retry_handlers=self.slack_retry_handlers,
+        )
+
+    @property
+    def _method_resolver(self):
+        if self.slack_method_version == "v1":
+            return self.slack_hook.send_file
+        return self.slack_hook.send_file_v1_to_v2
+
     def execute(self, context: Context) -> None:
         # Parse file format from filename
         output_file_format, _ = parse_filename(
             filename=self.slack_filename,
             supported_file_formats=self.SUPPORTED_FILE_FORMATS,
         )
 
-        slack_hook = SlackHook(
-            slack_conn_id=self.slack_conn_id,
-            base_url=self.slack_base_url,
-            timeout=self.slack_timeout,
-            proxy=self.slack_proxy,
-            retry_handlers=self.slack_retry_handlers,
-        )
         with NamedTemporaryFile(mode="w+", suffix=f"_{self.slack_filename}") as fp:
             # tempfile.NamedTemporaryFile used only for create and remove temporary file,
             # pandas will open file in correct mode itself depend on file type.
@@ -129,7 +145,7 @@ def execute(self, context: Context) -> None:
                 # if SUPPORTED_FILE_FORMATS extended and no actual implementation for specific format.
                 raise AirflowException(f"Unexpected output file format: {output_file_format}")
 
-            slack_hook.send_file(
+            self._method_resolver(
                 channels=self.slack_channels,
                 file=output_file_name,
                 filename=self.slack_filename,