Skip to content

Type-check the whole faust package and delete the mypy ratchet #758

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wbarnha merged 5 commits into from
Aug 6, 2026
Merged

Type-check the whole faust package and delete the mypy ratchet #758

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cd2729c
Make implicit Optional parameters explicit
claude Aug 5, 2026
fe4e23b
Correct the type-level contracts in faust/types
claude Aug 5, 2026
752d179
Type-check the rest of faust and delete the ratchet
claude Aug 6, 2026
b1767ce
Keep the aerospike type-ignores on the interpolating line
claude Aug 6, 2026
e1e250d
Require mode-streaming>=0.6.0 so the clean tree is reproducible
claude Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion faust/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,14 @@ class VersionInfo(NamedTuple):
if _match is None: # pragma: no cover
raise RuntimeError("THIS IS A BROKEN RELEASE!")
_temp = _match.groups()
VERSION = version_info = VersionInfo(*_temp)
# XXX This is broken and so is the public ``faust.version_info``: the regex
# yields the strings ``(prefix, version, suffix)``, which land positionally in
# the ``(major, minor, micro)`` int fields. So ``.major`` is the ``'v'``
# prefix or :const:`None`, ``.minor`` is the entire version string and
# ``.micro`` is the suffix -- e.g. ``VersionInfo(major=None,
# minor='0.11.5', micro='')`` instead of ``(0, 11, 5)``.
# Left as-is because fixing it changes what ``faust.VERSION`` holds.
VERSION = version_info = VersionInfo(*_temp) # type: ignore[arg-type]
del _match
del _temp
del re
Expand Down
92 changes: 46 additions & 46 deletions faust/agents/agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,15 +185,15 @@ def __init__(
*,
app: AppT,
name: Optional[str] = None,
channel: Union[str, ChannelT] = None,
channel: Optional[Union[str, ChannelT]] = None,
concurrency: int = 1,
sink: Iterable[SinkT] = None,
on_error: AgentErrorHandler = None,
supervisor_strategy: Type[SupervisorStrategyT] = None,
sink: Optional[Iterable[SinkT]] = None,
on_error: Optional[AgentErrorHandler] = None,
supervisor_strategy: Optional[Type[SupervisorStrategyT]] = None,
help: Optional[str] = None,
schema: Optional[SchemaT] = None,
key_type: ModelArg = None,
value_type: ModelArg = None,
key_type: Optional[ModelArg] = None,
value_type: Optional[ModelArg] = None,
isolated_partitions: bool = False,
use_reply_headers: Optional[bool] = None,
**kwargs: Any,
Expand Down Expand Up @@ -462,7 +462,7 @@ def info(self) -> Mapping:
"isolated_partitions": self.isolated_partitions,
}

def clone(self, *, cls: Type[AgentT] = None, **kwargs: Any) -> AgentT:
def clone(self, *, cls: Optional[Type[AgentT]] = None, **kwargs: Any) -> AgentT:
"""Create clone of this agent object.

Keyword arguments can be passed to override any argument
Expand All @@ -473,8 +473,8 @@ def clone(self, *, cls: Type[AgentT] = None, **kwargs: Any) -> AgentT:
def test_context(
self,
channel: Optional[ChannelT] = None,
supervisor_strategy: SupervisorStrategyT = None,
on_error: AgentErrorHandler = None,
supervisor_strategy: Optional[SupervisorStrategyT] = None,
on_error: Optional[AgentErrorHandler] = None,
**kwargs: Any,
) -> AgentTestWrapperT: # pragma: no cover
"""Create new unit-testing wrapper for this agent."""
Expand All @@ -501,11 +501,11 @@ async def on_agent_error(agent: AgentT, exc: BaseException) -> None:

def _prepare_channel(
self,
channel: Union[str, ChannelT] = None,
channel: Optional[Union[str, ChannelT]] = None,
internal: bool = True,
schema: Optional[SchemaT] = None,
key_type: ModelArg = None,
value_type: ModelArg = None,
key_type: Optional[ModelArg] = None,
value_type: Optional[ModelArg] = None,
**kwargs: Any,
) -> ChannelT:
app = self.app
Expand Down Expand Up @@ -768,12 +768,12 @@ def _response_class(self, value: Any) -> Type[ReqRepResponse]:

async def cast(
self,
value: V = None,
value: Optional[V] = None,
*,
key: K = None,
key: Optional[K] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
headers: Optional[HeadersArg] = None,
) -> None:
"""RPC operation: like :meth:`ask` but do not expect reply.

Expand All @@ -790,13 +790,13 @@ async def cast(

async def ask(
self,
value: V = None,
value: Optional[V] = None,
*,
key: K = None,
key: Optional[K] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
reply_to: ReplyToArg = None,
headers: Optional[HeadersArg] = None,
reply_to: Optional[ReplyToArg] = None,
correlation_id: Optional[str] = None,
) -> Any:
"""RPC operation: ask agent for result of processing value.
Expand All @@ -821,13 +821,13 @@ async def ask(

async def ask_nowait(
self,
value: V = None,
value: Optional[V] = None,
*,
key: K = None,
key: Optional[K] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
reply_to: ReplyToArg = None,
headers: Optional[HeadersArg] = None,
reply_to: Optional[ReplyToArg] = None,
correlation_id: Optional[str] = None,
force: bool = False,
) -> ReplyPromise:
Expand All @@ -853,11 +853,11 @@ async def ask_nowait(

def _create_req(
self,
key: K = None,
value: V = None,
reply_to: ReplyToArg = None,
key: Optional[K] = None,
value: Optional[V] = None,
reply_to: Optional[ReplyToArg] = None,
correlation_id: Optional[str] = None,
headers: HeadersArg = None,
headers: Optional[HeadersArg] = None,
) -> Tuple[V, Optional[HeadersArg]]:
if reply_to is None:
raise TypeError("Missing reply_to argument")
Expand Down Expand Up @@ -890,15 +890,15 @@ def _request_class(self, value: V) -> Type[ReqRepRequest]:
async def send(
self,
*,
key: K = None,
value: V = None,
key: Optional[K] = None,
value: Optional[V] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
key_serializer: CodecArg = None,
value_serializer: CodecArg = None,
headers: Optional[HeadersArg] = None,
key_serializer: Optional[CodecArg] = None,
value_serializer: Optional[CodecArg] = None,
callback: Optional[MessageSentCallback] = None,
reply_to: ReplyToArg = None,
reply_to: Optional[ReplyToArg] = None,
correlation_id: Optional[str] = None,
force: bool = False,
) -> Awaitable[RecordMetadata]:
Expand Down Expand Up @@ -930,8 +930,8 @@ def _get_strtopic(self, topic: Union[str, ChannelT, TopicT, AgentT]) -> str:
async def map(
self,
values: Union[AsyncIterable, Iterable],
key: K = None,
reply_to: ReplyToArg = None,
key: Optional[K] = None,
reply_to: Optional[ReplyToArg] = None,
) -> AsyncIterator: # pragma: no cover
"""RPC map operation on a list of values.

Expand All @@ -947,7 +947,7 @@ async def map(
async def kvmap(
self,
items: Union[AsyncIterable[Tuple[K, V]], Iterable[Tuple[K, V]]],
reply_to: ReplyToArg = None,
reply_to: Optional[ReplyToArg] = None,
) -> AsyncIterator[str]: # pragma: no cover
"""RPC map operation on a list of ``(key, value)`` pairs.

Expand Down Expand Up @@ -980,8 +980,8 @@ async def kvmap(
async def join(
self,
values: Union[AsyncIterable[V], Iterable[V]],
key: K = None,
reply_to: ReplyToArg = None,
key: Optional[K] = None,
reply_to: Optional[ReplyToArg] = None,
) -> List[Any]: # pragma: no cover
"""RPC map operation on a list of values.

Expand All @@ -996,7 +996,7 @@ async def join(
async def kvjoin(
self,
items: Union[AsyncIterable[Tuple[K, V]], Iterable[Tuple[K, V]]],
reply_to: ReplyToArg = None,
reply_to: Optional[ReplyToArg] = None,
) -> List[Any]: # pragma: no cover
"""RPC map operation on list of ``(key, value)`` pairs.

Expand Down Expand Up @@ -1159,15 +1159,15 @@ async def crash_test_agent(self, exc: BaseException) -> None:

async def put(
self,
value: V = None,
key: K = None,
value: Optional[V] = None,
key: Optional[K] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
key_serializer: CodecArg = None,
value_serializer: CodecArg = None,
headers: Optional[HeadersArg] = None,
key_serializer: Optional[CodecArg] = None,
value_serializer: Optional[CodecArg] = None,
*,
reply_to: ReplyToArg = None,
reply_to: Optional[ReplyToArg] = None,
correlation_id: Optional[str] = None,
wait: bool = True,
) -> EventT:
Expand Down Expand Up @@ -1203,7 +1203,7 @@ def to_message(
offset: int = 0,
timestamp: Optional[float] = None,
timestamp_type: int = 0,
headers: HeadersArg = None,
headers: Optional[HeadersArg] = None,
) -> Message:
try:
topic_name = self._get_strtopic(self.original_channel)
Expand Down
25 changes: 16 additions & 9 deletions faust/app/_attached.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,14 +90,14 @@ def enabled(self) -> bool:
async def maybe_put(
self,
channel: Union[ChannelT, str],
key: K = None,
value: V = None,
key: Optional[K] = None,
value: Optional[V] = None,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
headers: Optional[HeadersArg] = None,
schema: Optional[SchemaT] = None,
key_serializer: CodecArg = None,
value_serializer: CodecArg = None,
key_serializer: Optional[CodecArg] = None,
value_serializer: Optional[CodecArg] = None,
callback: Optional[MessageSentCallback] = None,
force: bool = False,
) -> Awaitable[RecordMetadata]:
Expand Down Expand Up @@ -148,10 +148,10 @@ def put(
value: V,
partition: Optional[int] = None,
timestamp: Optional[float] = None,
headers: HeadersArg = None,
headers: Optional[HeadersArg] = None,
schema: Optional[SchemaT] = None,
key_serializer: CodecArg = None,
value_serializer: CodecArg = None,
key_serializer: Optional[CodecArg] = None,
value_serializer: Optional[CodecArg] = None,
callback: Optional[MessageSentCallback] = None,
) -> Awaitable[RecordMetadata]:
"""Attach message to source topic offset."""
Expand Down Expand Up @@ -183,8 +183,15 @@ def put(

async def commit(self, tp: TP, offset: int) -> None:
"""Publish all messaged attached to topic partition and offset."""
# XXX ``publish_for_tp_offset`` is typed as returning bare awaitables,
# but ``asyncio.wait`` accepts only futures/tasks -- passing a plain
# coroutine raises TypeError on Python 3.11+. In-tree channels always
# hand back an ``asyncio.Future``, so this works today; a third-party
# ``ChannelT.publish_message`` returning a coroutine would break it.
# Wrapping in ``ensure_future`` here would change runtime behaviour,
# so the error is silenced instead.
await asyncio.wait(
await self.publish_for_tp_offset(tp, offset),
await self.publish_for_tp_offset(tp, offset), # type: ignore[type-var]
return_when=asyncio.ALL_COMPLETED,
)

Expand Down
Loading
Loading