diff --git a/src/datastar_py/django.py b/src/datastar_py/django.py index 955a343..bda7ea6 100644 --- a/src/datastar_py/django.py +++ b/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]: diff --git a/src/datastar_py/sse.py b/src/datastar_py/sse.py index 94124c2..ccf5081 100644 --- a/src/datastar_py/sse.py +++ b/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: diff --git a/tests/test_django_decorator_typing.py b/tests/test_django_decorator_typing.py new file mode 100644 index 0000000..96cbc5a --- /dev/null +++ b/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 diff --git a/tests/typing_fixtures/django_decorator.py b/tests/typing_fixtures/django_decorator.py new file mode 100644 index 0000000..16e66ab --- /dev/null +++ b/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('
') + + +@datastar_response +async def async_coro_view(request: HttpRequest) -> SyncDatastarEvents: + return SSE.patch_elements('
') + + +@datastar_response +async def async_gen_view(request: HttpRequest) -> AsyncDatastarEvents: + yield SSE.patch_elements('
') + + +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), +]