Fix @datastar_response typing for Django async views

src/datastar_py/django.py

@@ -1,15 +1,22 @@
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Coroutine, Mapping
 from functools import wraps
 from inspect import isasyncgenfunction, isawaitable, iscoroutinefunction
-from typing import Any, ParamSpec
+from typing import Any, ParamSpec, overload
 
 from django.http import HttpRequest
 from django.http import StreamingHttpResponse as _StreamingHttpResponse
 
 from . import _read_signals
-from .sse import SSE_HEADERS, DatastarEvent, DatastarEvents, ServerSentEventGenerator
+from .sse import (
+    SSE_HEADERS,
+    AsyncDatastarEvents,
+    DatastarEvent,
+    DatastarEvents,
+    ServerSentEventGenerator,
+    SyncDatastarEvents,
+)
 
 __all__ = [
     "SSE_HEADERS",
@@ -44,6 +51,24 @@ def __init__(
 P = ParamSpec("P")
 
 
+@overload
+def datastar_response(
+    func: Callable[P, Coroutine[Any, Any, SyncDatastarEvents]],
+) -> Callable[P, Coroutine[Any, Any, DatastarResponse]]: ...
+
+
+@overload
+def datastar_response(
+    func: Callable[P, AsyncDatastarEvents],
+) -> Callable[P, Coroutine[Any, Any, DatastarResponse]]: ...
+
+
+@overload
+def datastar_response(
+    func: Callable[P, SyncDatastarEvents],
+) -> Callable[P, DatastarResponse]: ...
+
+
 def datastar_response(
     func: Callable[P, Awaitable[DatastarEvents] | DatastarEvents],
 ) -> Callable[P, Awaitable[DatastarResponse] | DatastarResponse]:

src/datastar_py/sse.py

@@ -32,9 +32,9 @@ class DatastarEvent(str):
 
 
 # 0..N datastar events
-DatastarEvents: TypeAlias = (
-    DatastarEvent | Iterable[DatastarEvent] | AsyncIterable[DatastarEvent] | None
-)
+SyncDatastarEvents: TypeAlias = DatastarEvent | Iterable[DatastarEvent] | None
+AsyncDatastarEvents: TypeAlias = AsyncIterable[DatastarEvent]
+DatastarEvents: TypeAlias = SyncDatastarEvents | AsyncDatastarEvents
 
 
 class ServerSentEventGenerator:

tests/test_django_decorator_typing.py

@@ -0,0 +1,41 @@
+"""Static typing regression tests for Django datastar_response overloads."""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+FIXTURE_PATH = Path(__file__).parent / "typing_fixtures" / "django_decorator.py"
+
+
+@pytest.mark.skipif(importlib.util.find_spec("mypy") is None, reason="mypy not installed")
+@pytest.mark.skipif(importlib.util.find_spec("django") is None, reason="django not installed")
+def test_django_datastar_response_mypy_overloads() -> None:
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "mypy",
+            "--hide-error-context",
+            "--no-error-summary",
+            str(FIXTURE_PATH),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    output = result.stdout + result.stderr
+
+    assert result.returncode == 0, output
+    assert (
+        'Revealed type is "def (request: django.http.request.HttpRequest) '
+        '-> datastar_py.django.DatastarResponse"'
+    ) in output
+    assert output.count(
+        'Revealed type is "def (request: django.http.request.HttpRequest) '
+        '-> typing.Coroutine[Any, Any, datastar_py.django.DatastarResponse]"'
+    ) == 2

tests/typing_fixtures/django_decorator.py

@@ -0,0 +1,39 @@
+from collections.abc import Callable, Coroutine
+from typing import Any, reveal_type
+
+from django.http import HttpRequest
+from django.urls import path
+
+from datastar_py.django import DatastarResponse, datastar_response
+from datastar_py.sse import AsyncDatastarEvents, SyncDatastarEvents
+from datastar_py.sse import ServerSentEventGenerator as SSE
+
+
+@datastar_response
+def sync_view(request: HttpRequest) -> SyncDatastarEvents:
+    return SSE.patch_elements('<div id="sync"></div>')
+
+
+@datastar_response
+async def async_coro_view(request: HttpRequest) -> SyncDatastarEvents:
+    return SSE.patch_elements('<div id="coro"></div>')
+
+
+@datastar_response
+async def async_gen_view(request: HttpRequest) -> AsyncDatastarEvents:
+    yield SSE.patch_elements('<div id="gen"></div>')
+
+
+sync_typed: Callable[[HttpRequest], DatastarResponse] = sync_view
+async_coro_typed: Callable[[HttpRequest], Coroutine[Any, Any, DatastarResponse]] = async_coro_view
+async_gen_typed: Callable[[HttpRequest], Coroutine[Any, Any, DatastarResponse]] = async_gen_view
+
+reveal_type(sync_view)
+reveal_type(async_coro_view)
+reveal_type(async_gen_view)
+
+urlpatterns = [
+    path("sync/", sync_view),
+    path("coro/", async_coro_view),
+    path("gen/", async_gen_view),
+]