adopt async_dispatch for slack collection

src/integrations/prefect-slack/prefect_slack/credentials.py

@@ -1,14 +1,15 @@
 """Credential classes use to store Slack credentials."""
 
-from typing import Optional
+from typing import Any, Optional, Union
 
 from pydantic import Field, SecretStr
 from slack_sdk.web.async_client import AsyncWebClient
 from slack_sdk.webhook.async_client import AsyncWebhookClient
+from slack_sdk.webhook.client import WebhookClient
 
+from prefect._internal.compatibility.async_dispatch import async_dispatch
 from prefect.blocks.core import Block
 from prefect.blocks.notifications import NotificationBlock
-from prefect.utilities.asyncutils import sync_compatible
 
 
 class SlackCredentials(Block):
@@ -49,6 +50,14 @@ def get_client(self) -> AsyncWebClient:
         return AsyncWebClient(token=self.token.get_secret_value())
 
 
+async def _notify_async(obj: Any, body: str, subject: Optional[str] = None):
+    client = obj.get_client()
+
+    response = await client.send(text=body)
+
+    obj._raise_on_failure(response)
+
+
 class SlackWebhook(NotificationBlock):
     """
     Block holding a Slack webhook for use in tasks and flows.
@@ -90,22 +99,18 @@ class SlackWebhook(NotificationBlock):
         examples=["https://hooks.slack.com/XXX"],
     )
 
-    def get_client(self) -> AsyncWebhookClient:
+    def get_client(
+        self, sync_client: bool = False
+    ) -> Union[AsyncWebhookClient, WebhookClient]:
         """
         Returns an authenticated `AsyncWebhookClient` to interact with the configured
         Slack webhook.
         """
+        if sync_client:
+            return WebhookClient(url=self.url.get_secret_value())
         return AsyncWebhookClient(url=self.url.get_secret_value())
 
-    @sync_compatible
-    async def notify(self, body: str, subject: Optional[str] = None):
-        """
-        Sends a message to the Slack channel.
-        """
-        client = self.get_client()
-
-        response = await client.send(text=body)
-
+    def _raise_on_failure(self, response: Any):
         # prefect>=2.17.2 added a means for notification blocks to raise errors on
         # failures. This is not available in older versions, so we need to check if the
         # private base class attribute exists before using it.
@@ -117,3 +122,20 @@ async def notify(self, body: str, subject: Optional[str] = None):
 
             if response.status_code >= 400:
                 raise NotificationError(f"Failed to send message: {response.body}")
+
+    async def notify_async(self, body: str, subject: Optional[str] = None):
+        """
+        Sends a message to the Slack channel.
+        """
+        await _notify_async(self, body, subject)
+
+    @async_dispatch(_notify_async)  # type: ignore
+    def notify(self, body: str, subject: Optional[str] = None):
+        """
+        Sends a message to the Slack channel.
+        """
+        client = self.get_client(sync_client=True)
+
+        response = client.send(text=body)
+
+        self._raise_on_failure(response)

src/integrations/prefect-slack/pyproject.toml

@@ -22,11 +22,7 @@ classifiers = [
   "Programming Language :: Python :: 3.12",
   "Topic :: Software Development :: Libraries",
 ]
-dependencies = [
-  "aiohttp",
-  "slack_sdk>=3.15.1",
-  "prefect>=3.0.0rc1",
-]
+dependencies = ["aiohttp", "slack_sdk>=3.15.1", "prefect>=3.0.0rc1"]
 dynamic = ["version"]
 
 [project.optional-dependencies]
@@ -74,7 +70,6 @@ fail_under = 80
 show_missing = true
 
 [tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
 asyncio_mode = "auto"
-env = [
-  "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]

src/integrations/prefect-slack/tests/test_credentials.py

@@ -1,9 +1,10 @@
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from prefect_slack import SlackCredentials, SlackWebhook
 from slack_sdk.web.async_client import AsyncWebClient
-from slack_sdk.webhook.async_client import AsyncWebhookClient, WebhookResponse
+from slack_sdk.webhook.async_client import AsyncWebhookClient
+from slack_sdk.webhook.webhook_response import WebhookResponse
 
 
 def test_slack_credentials():
@@ -63,3 +64,52 @@ async def test_slack_webhook_block_handles_raise_on_failure(
     with pytest.raises(NotificationError, match="Failed to send message: woops"):
         with block.raise_on_failure():
             await block.notify("hello", "world")
+
+
+def test_slack_webhook_sync_notify(monkeypatch):
+    """Test the sync notify path"""
+    mock_client = MagicMock()
+    mock_client.send.return_value = WebhookResponse(
+        url="http://test", status_code=200, body="ok", headers={}
+    )
+
+    webhook = SlackWebhook(url="http://test")
+    monkeypatch.setattr(webhook, "get_client", MagicMock(return_value=mock_client))
+
+    webhook.notify("test message")
+    mock_client.send.assert_called_once_with(text="test message")
+
+
+async def test_slack_webhook_async_notify(monkeypatch):
+    """Test the async notify path"""
+    mock_client = MagicMock()
+    mock_client.send = AsyncMock(
+        return_value=WebhookResponse(
+            url="http://test", status_code=200, body="ok", headers={}
+        )
+    )
+
+    webhook = SlackWebhook(url="http://test")
+    monkeypatch.setattr(webhook, "get_client", MagicMock(return_value=mock_client))
+
+    await webhook.notify_async("test message")
+    mock_client.send.assert_called_once_with(text="test message")
+
+
+@pytest.mark.parametrize("message", ["test message 1", "test message 2"])
+async def test_slack_webhook_notify_async_dispatch(monkeypatch, message):
+    """Test that async_dispatch properly handles both sync and async contexts"""
+
+    mock_response = WebhookResponse(
+        url="http://test", status_code=200, body="ok", headers={}
+    )
+
+    mock_client = MagicMock()
+    mock_client.send = AsyncMock(return_value=mock_response)
+
+    webhook = SlackWebhook(url="http://test")
+    monkeypatch.setattr(webhook, "get_client", lambda sync_client=False: mock_client)
+
+    # Test notification
+    await webhook.notify(message)
+    mock_client.send.assert_called_once_with(text=message)