workflow_streams: make topic(type=...) optional, default to Any

Per discussion: heterogeneous topics and dynamic-topic forwarders
previously had to write
type=cast(type[Any], cast(object, Any)) at the call site to
satisfy pyright (typing.Any is a special form, not a class). Make
the type kwarg optional and default to typing.Any so the natural
form is just client.topic("name") / stream.topic("name").

The type-uniformity invariant is unchanged: each instance binds a
topic name to exactly one type; mixing untyped (= Any) with a
specific type still raises. typing.Any can also be passed
explicitly, with the cast still required for type-strict callers.

Adds overloads so callers that pass type=T still get a typed
TopicHandle[T] from pyright; the no-type form returns
TopicHandle[Any] / WorkflowTopicHandle[Any]. Updates the existing
test to exercise both the omitted-type and explicit-type=Any
paths.

Author: jssmith

temporalio/contrib/workflow_streams/_client.py

@@ -23,7 +23,7 @@
 import uuid
 from collections.abc import AsyncIterator
 from datetime import timedelta
-from typing import Any, TypeVar
+from typing import Any, TypeVar, overload
 
 from typing_extensions import Self
 
@@ -246,12 +246,19 @@ def _publish_to_topic(
         ):
             self._flush_event.set()
 
-    def topic(self, name: str, *, type: type[T]) -> TopicHandle[T]:
+    @overload
+    def topic(self, name: str) -> TopicHandle[Any]: ...
+    @overload
+    def topic(self, name: str, *, type: type[T]) -> TopicHandle[T]: ...
+
+    def topic(
+        self, name: str, *, type: type[T] | None = None
+    ) -> TopicHandle[T] | TopicHandle[Any]:
         """Return a typed handle for publishing to and subscribing from ``name``.
 
         The handle records the topic name and value type so call sites
         do not have to repeat them. Each :class:`WorkflowStreamClient`
-        instance binds a topic name to exactly one ``T``: a second call
+        instance binds a topic name to exactly one type: a second call
         with an unequal type raises ``RuntimeError``. Repeating the
         same call with the same type is idempotent and returns an
         equivalent handle.
@@ -261,46 +268,53 @@ def topic(self, name: str, *, type: type[T]) -> TopicHandle[T]:
         equality on the type object; subtype and union-superset
         relationships are not recognized.
 
-        For heterogeneous topics or dynamic-topic forwarders, pass
-        ``type=typing.Any``; subscribers receive the converter's
-        default decoded value. Pre-built ``Payload`` values can be
-        passed to :meth:`TopicHandle.publish` regardless of the bound
-        type (zero-copy fast path) — there is no need to bind the
-        topic to ``Payload`` itself, and doing so would break the
-        subscribe path (use ``result_type=RawValue`` on
-        :meth:`subscribe` if you need raw payloads on a subscriber).
+        Omitting ``type`` (or passing ``type=typing.Any``) is the
+        documented escape hatch for heterogeneous topics or
+        dynamic-topic forwarders: the handle accepts any value, and
+        subscribers receive the converter's default decoded value.
+        Pre-built ``Payload`` values can be passed to
+        :meth:`TopicHandle.publish` regardless of the bound type
+        (zero-copy fast path) — there is no need to bind the topic to
+        ``Payload`` itself, and doing so would break the subscribe
+        path (use ``result_type=RawValue`` on
+        :meth:`WorkflowStreamClient.subscribe` if you need raw
+        payloads on a subscriber).
 
         Args:
             name: Topic name.
             type: Value type bound to this handle. Used as the
                 ``result_type`` when subscribing through the handle.
+                Defaults to ``typing.Any`` (heterogeneous topic).
 
         Returns:
-            :class:`TopicHandle` bound to ``name`` and ``type``.
+            :class:`TopicHandle` bound to ``name`` and the resolved
+            type.
 
         Raises:
             RuntimeError: If ``name`` is already bound on this client
                 to a different type.
         """
-        if type is Payload:
+        bound: Any = Any if type is None else type
+        if bound is Payload:
             raise RuntimeError(
                 "Cannot bind a topic to type=Payload: the payload converter "
                 "has no Payload decode path, so TopicHandle.subscribe would "
                 "fail. Pre-built Payload values can be passed to "
                 "TopicHandle.publish on any-typed handle (zero-copy fast "
-                "path); use type=typing.Any for heterogeneous topics, and "
-                "subscribe via WorkflowStreamClient.subscribe with "
-                "result_type=RawValue when raw payloads are needed."
+                "path); omit type (or pass type=typing.Any) for "
+                "heterogeneous topics, and subscribe via "
+                "WorkflowStreamClient.subscribe with result_type=RawValue "
+                "when raw payloads are needed."
             )
         existing = self._topic_types.get(name)
-        if existing is not None and existing != type:
+        if existing is not None and existing != bound:
             raise RuntimeError(
                 f"Topic {name!r} is already bound to type {existing!r} on this "
-                f"client; refusing to rebind to {type!r}. Use a single type "
-                f"per topic, or pass type=typing.Any for heterogeneous topics."
+                f"client; refusing to rebind to {bound!r}. Use a single type "
+                f"per topic, or omit type (=typing.Any) for heterogeneous topics."
             )
-        self._topic_types[name] = type
-        return TopicHandle(self, name, type)
+        self._topic_types[name] = bound
+        return TopicHandle(self, name, bound)
 
     async def flush(self) -> None:
         """Flush buffered (and pending) items and wait for server confirmation.

temporalio/contrib/workflow_streams/_stream.py

@@ -29,7 +29,7 @@
 import sys
 from collections.abc import Sequence
 from datetime import timedelta
-from typing import Any, Callable, NoReturn, TypeVar
+from typing import Any, Callable, NoReturn, TypeVar, overload
 
 from temporalio import workflow
 from temporalio.api.common.v1 import Payload
@@ -170,14 +170,21 @@ def _publish_to_topic(self, topic: str, value: Any) -> None:
             payload = workflow.payload_converter().to_payloads([value])[0]
         self._log.append(WorkflowStreamItem(topic=topic, data=payload))
 
-    def topic(self, name: str, *, type: type[T]) -> WorkflowTopicHandle[T]:
+    @overload
+    def topic(self, name: str) -> WorkflowTopicHandle[Any]: ...
+    @overload
+    def topic(self, name: str, *, type: type[T]) -> WorkflowTopicHandle[T]: ...
+
+    def topic(
+        self, name: str, *, type: type[T] | None = None
+    ) -> WorkflowTopicHandle[T] | WorkflowTopicHandle[Any]:
         """Return a typed handle for publishing to ``name`` from this workflow.
 
         The handle records the topic name and value type so call sites
         do not have to repeat them. Each :class:`WorkflowStream`
-        instance binds a topic name to exactly one ``T``: a second
-        call with an unequal type raises ``RuntimeError``. Repeating
-        the same call with the same type is idempotent and returns an
+        instance binds a topic name to exactly one type: a second call
+        with an unequal type raises ``RuntimeError``. Repeating the
+        same call with the same type is idempotent and returns an
         equivalent handle.
 
         Type uniformity is checked only on this stream instance — it
@@ -186,40 +193,44 @@ def topic(self, name: str, *, type: type[T]) -> WorkflowTopicHandle[T]:
         on the type object; subtype and union-superset relationships
         are not recognized.
 
-        For heterogeneous topics, pass ``type=typing.Any``. Pre-built
+        Omitting ``type`` (or passing ``type=typing.Any``) is the
+        documented escape hatch for heterogeneous topics. Pre-built
         ``Payload`` values can be passed to
         :meth:`WorkflowTopicHandle.publish` regardless of the bound
         type (zero-copy fast path) — there is no need to bind the
         topic to ``Payload`` itself.
 
         Args:
             name: Topic name.
-            type: Value type bound to this handle.
+            type: Value type bound to this handle. Defaults to
+                ``typing.Any`` (heterogeneous topic).
 
         Returns:
-            :class:`WorkflowTopicHandle` bound to ``name`` and ``type``.
+            :class:`WorkflowTopicHandle` bound to ``name`` and the
+            resolved type.
 
         Raises:
             RuntimeError: If ``name`` is already bound on this stream
                 to a different type.
         """
-        if type is Payload:
+        bound: Any = Any if type is None else type
+        if bound is Payload:
             raise RuntimeError(
                 "Cannot bind a topic to type=Payload. Pre-built Payload "
                 "values can be passed to WorkflowTopicHandle.publish on "
-                "any-typed handle (zero-copy fast path); use "
-                "type=typing.Any for heterogeneous topics."
+                "any-typed handle (zero-copy fast path); omit type (or "
+                "pass type=typing.Any) for heterogeneous topics."
             )
         existing = self._topic_types.get(name)
-        if existing is not None and existing != type:
+        if existing is not None and existing != bound:
             raise RuntimeError(
                 f"Topic {name!r} is already bound to type {existing!r} on this "
-                f"workflow stream; refusing to rebind to {type!r}. Use a "
-                f"single type per topic, or pass type=typing.Any for "
+                f"workflow stream; refusing to rebind to {bound!r}. Use a "
+                f"single type per topic, or omit type (=typing.Any) for "
                 f"heterogeneous topics."
             )
-        self._topic_types[name] = type
-        return WorkflowTopicHandle(self, name, type)
+        self._topic_types[name] = bound
+        return WorkflowTopicHandle(self, name, bound)
 
     def get_state(
         self, *, publisher_ttl: timedelta = timedelta(seconds=900)

tests/contrib/workflow_streams/test_workflow_streams.py

@@ -656,12 +656,16 @@ async def test_topic_handle_client_uniqueness(client: Client) -> None:
     other = stream.topic("other", type=bytes)
     assert other.type is bytes
 
-    # Any escape hatch coexists on a different topic. ``Any`` is a
-    # typing special form, not a class, so cast through ``object`` to
-    # satisfy ``type[T]`` — the runtime check uses Python equality and
-    # accepts ``Any`` directly.
-    raw = stream.topic("forwarded", type=cast(type[Any], cast(object, Any)))
+    # Any escape hatch coexists on a different topic. Omitting ``type``
+    # is the documented form (defaults to ``typing.Any``); we also
+    # exercise the explicit ``type=Any`` path with the cast required
+    # because ``Any`` is a typing special form rather than a class.
+    raw = stream.topic("forwarded")
     assert raw.type is Any
+    explicit = stream.topic(
+        "forwarded-explicit", type=cast(type[Any], cast(object, Any))
+    )
+    assert explicit.type is Any
 
     # Binding to Payload itself is rejected — subscribers would have
     # no decode path. Pre-built Payload values can still be published