From 3ab958b8ef88fb914230e29aa138286c082e9a2b Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Thu, 3 Sep 2026 19:08:47 -0500 Subject: [PATCH 01/14] init cust elaboration plug tests Signed-off-by: Dibri Nsofor --- test-data/unit/check-elaboration-plugin.test | 39 ++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 test-data/unit/check-elaboration-plugin.test diff --git a/test-data/unit/check-elaboration-plugin.test b/test-data/unit/check-elaboration-plugin.test new file mode 100644 index 000000000000..c52c3530d902 --- /dev/null +++ b/test-data/unit/check-elaboration-plugin.test @@ -0,0 +1,39 @@ +[case testElaboratesAnyReturn] +from typing import Any + +def bad(x: int) -> Any: + return str(x) + +reveal_type(bad(1)) # N: Revealed type is "builtins.str" +value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") + +[case testElaboratesIgnoresBadAnnot] +from typing import Any + +def unused(x: int) -> Any: + return "hello" + +reveal_type(unused(1)) # N: Revealed type is "builtins.str" +value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") + +[case testElaboratesAttribute] +from typing import Any + +class Settings: + retries: Any = "3" + +reveal_type(Settings().retries) # N: Revealed type is "builtins.str" +retries: int = Settings().retries # E: Incompatible types in assignment (expression has type "str", variable has type "int") + + +[case testElaboratesNestedCallableReturn] +from typing import Any, Callable + +handlers: dict[str, Callable[[bytes], Any]] = { + "decode": lambda data: "decoded" +} + +result = handlers["decode"](b"hello") + +reveal_type(result) # N: Revealed type is "builtins.str" +value: bytes = result # E: Incompatible types in assignment (expression has type "str", variable has type "bytes") \ No newline at end of file From cd30bb5e878e399e5d4de7ff81662295e44f0015 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Fri, 4 Sep 2026 19:22:13 -0500 Subject: [PATCH 02/14] tests for use case from #13668 Signed-off-by: Dibri Nsofor --- test-data/unit/check-union-rule-enforced.test | 182 ++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 test-data/unit/check-union-rule-enforced.test diff --git a/test-data/unit/check-union-rule-enforced.test b/test-data/unit/check-union-rule-enforced.test new file mode 100644 index 000000000000..f23e51e7b263 --- /dev/null +++ b/test-data/unit/check-union-rule-enforced.test @@ -0,0 +1,182 @@ +[case testUnionRuleReportsOnceAtDefinition] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def choose(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +def first_caller() -> None: + choose(1) + + +def second_caller() -> None: + choose("hello") + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testUnionRuleRunsWithoutAnyCalls] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def never_called(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testUnionRuleCoversMethods] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +class Parser: + def parse(self, value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +Parser().parse(1) +Parser().parse("hello") + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testUnionRuleCanBeIgnoredAtDefinition] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def permitted(value: int | str) -> int | str: # type: ignore[union-return] + return value + + +permitted(1) +permitted("hello") + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testUnionRuleRunsOnceForConstrainedTypeVar] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import TypeVar + +T = TypeVar("T", int, str) + + +def choose(value: T) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testAnyReturnRefinesCallType] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def bad(x: int) -> Any: + return str(x) + + +reveal_type(bad(1)) # N: Revealed type is "builtins.str" + +value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testAnyReturnRefinesEarlierAndLaterCallers] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def before_definition() -> None: + reveal_type(bad(1)) # N: Revealed type is "builtins.str" + value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +def bad(x: int) -> Any: + return str(x) + + +def after_definition() -> None: + reveal_type(bad(1)) # N: Revealed type is "builtins.str" + value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testAnyReturnRefinementReachesFixedPoint] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def use_result() -> None: + reveal_type(first()) # N: Revealed type is "builtins.str" + value: int = first() # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +def first() -> Any: + return second() + + +def second() -> Any: + return "done" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testAnyReturnRuleDoesNotChangeNonAnyAnnotation] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def returns_object() -> object: + return "hello" + + +reveal_type(returns_object()) # N: Revealed type is "builtins.object" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + + +[case testAnyReturnRuleLeavesBodyAnyAlone] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def identity(value: Any) -> Any: + return value + + +reveal_type(identity(1)) # N: Revealed type is "Any" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py From 420aa866991fab82c1f6e71ca5806264440e4608 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Fri, 4 Sep 2026 20:59:48 -0500 Subject: [PATCH 03/14] boiler struct for api ext Signed-off-by: Dibri Nsofor --- mypy/plugin.py | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/mypy/plugin.py b/mypy/plugin.py index d0e5add71a5b..d2514bea78b6 100644 --- a/mypy/plugin.py +++ b/mypy/plugin.py @@ -137,6 +137,8 @@ class C: pass MypyFile, SymbolTableNode, TypeInfo, + FuncDef, + ReturnStmt ) from mypy.options import Options from mypy.types import ( @@ -516,6 +518,35 @@ class DynamicClassDefContext(NamedTuple): name: str # The name this class is being assigned to api: SemanticAnalyzerPluginInterface +# A context for a function hook signature after available after decl. +class FunctionDefContext(NamedTuple): + definition: FuncDef + declared_signature: CallableType + api: SemanticAnalyzerPluginInterface + +class ReturnSite(NamedTuple): + statement: ReturnStmt + inferred_type: Type + +# A context for a function hook after the body is checked. +class FunctionBodyContext(NamedTuple): + definition: FuncDef + declared_signature: CallableType # preserves the user annotations + inferred_return_type: Type # inferred from return exprs in func, can be None + return_sites: tuple[ReturnSite, ...] + can_fall_through: bool # tracks definitions with multiple exists (e.g early returns) + api: CheckerPluginInterface + +# None here means inferred function return type is not propagated +class FunctionBodyResult(NamedTuple): + refined_return: Type | None = None + +# Supplying this callback tells mypy that this body's inferred information may affect the function's published signature. +class FunctionDefHookResult(NamedTuple): + after_body: FunctionBodyHook | None = None + +FunctionBodyHook = Callable[[FunctionBodyContext], FunctionBodyResult] +FunctionDefHook = Callable[[FunctionDefContext], FunctionDefHookResult] @mypyc_attr(allow_interpreted_subclasses=True) class Plugin(CommonPluginApi): @@ -819,6 +850,14 @@ def get_dynamic_class_hook( """ return None + def get_function_def_hook( + self, fullname: str + ) -> FunctionDefHook: + """FILL + + """ + return None + T = TypeVar("T") From 3288ae8a512eff03e5b98f17c7e8fbe9d77ef25d Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Fri, 4 Sep 2026 21:58:47 -0500 Subject: [PATCH 04/14] threading through chain api Signed-off-by: Dibri Nsofor --- mypy/plugin.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/mypy/plugin.py b/mypy/plugin.py index d2514bea78b6..e75815472062 100644 --- a/mypy/plugin.py +++ b/mypy/plugin.py @@ -923,6 +923,9 @@ def get_function_signature_hook( def get_function_hook(self, fullname: str) -> Callable[[FunctionContext], Type] | None: return self._find_hook(lambda plugin: plugin.get_function_hook(fullname)) + def get_function_def_hook(self, fullname:str) -> Callable[[FunctionDefContext], FunctionDefHookResult] | None: + return self._find_hook(lambda plugin: plugin.get_function_def_hook(fullname)) + def get_method_signature_hook( self, fullname: str ) -> Callable[[MethodSigContext], FunctionLike] | None: From 725c622b43482cdce86d98952178fcc62b41bb7c Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Sat, 5 Sep 2026 23:43:43 -0500 Subject: [PATCH 05/14] prepass + tracking refined types Signed-off-by: Dibri Nsofor --- mypy/checker.py | 121 +++++++++++++++++++++++++++++++++++++++++++--- mypy/nodes.py | 15 +++--- mypy/plugin.py | 36 +++++++++----- mypy/traverser.py | 32 ++++++++++++ 4 files changed, 180 insertions(+), 24 deletions(-) diff --git a/mypy/checker.py b/mypy/checker.py index 33ed5387554d..4bafea7e4c5c 100644 --- a/mypy/checker.py +++ b/mypy/checker.py @@ -212,7 +212,13 @@ def __init__(self) -> None: from mypy.operators import flip_ops, int_op_to_method, neg_ops from mypy.options import PRECISE_TUPLE_TYPES, Options from mypy.patterns import AsPattern, StarredPattern -from mypy.plugin import Plugin +from mypy.plugin import ( + FunctionBodyContext, + FunctionBodyHook, + FunctionDefContext, + Plugin, + ReturnSite, +) from mypy.plugins import dataclasses as dataclasses_plugin from mypy.scope import Scope from mypy.semanal import is_trivial_body, refers_to_fullname, set_callable_name @@ -232,7 +238,12 @@ def __init__(self) -> None: restrict_subtype_away, unify_generic_callable, ) -from mypy.traverser import TraverserVisitor, all_return_statements, has_return_statement +from mypy.traverser import ( + TraverserVisitor, + all_return_statements, + can_fall_through, + has_return_statement, +) from mypy.treetransform import TransformVisitor from mypy.typeanal import check_for_explicit_any, has_any_from_unimported_type, make_optional_type from mypy.typeops import ( @@ -334,6 +345,14 @@ class FineGrainedDeferredNode(NamedTuple): active_typeinfo: TypeInfo | None +# Solely a plugin extension that tracks extra info internally while mypy visits a function body +class ActiveFunctionHook(NamedTuple): + defn: FuncDef + declared_signature: CallableType + callback: FunctionBodyHook + return_sites: list[ReturnSite] + + # Data structure returned by find_isinstance_check representing # information learned from the truth or falsehood of a condition. The # dict maps nodes representing expressions like 'a[0].x' to their @@ -461,7 +480,10 @@ class TypeChecker(NodeVisitor[None], TypeCheckerSharedApi, SplittingVisitor): # Plugin that provides special type checking rules for specific library # functions such as open(), etc. + # and private attributes to support elaboration plugin: Plugin + _function_body_hooks: list[ActiveFunctionHook] + _changed_plugin_signatures: set[str] # A helper state to produce unique temporary names on demand. _unique_id: int @@ -512,6 +534,8 @@ def __init__( self.inferred_attribute_types = None self.allow_constructor_cache = True self.local_type_map = LocalTypeMap(self) + self._function_body_hooks = [] + self._changed_plugin_signatures = set() self.can_skip_diagnostics: Final = ( self.options.ignore_errors @@ -1333,6 +1357,81 @@ def get_generator_return_type(self, return_type: Type, is_coroutine: bool) -> Ty # Treat `Iterator[X]` as a shorthand for `Generator[X, Any, None]`. return NoneType() + def record_plugin_return(self, statement: ReturnStmt, inferred_type: Type) -> None: + if not self._function_body_hooks: + return + + active = self._function_body_hooks[-1] + + # TODO: add test case + if self.scope.current_function() is not active.definition: + return + + active.return_sites.append( + ReturnSite( + statement=statement, + inferred_type=inferred_type, + ) + ) + + def publish_plugin_refinement( + self, defn: FuncDef, declared: CallableType, refined_return: Type + ) -> None: + new_signature = declared.copy_modified(ret_type=refined_return) + + old_signature = defn.plugin_effective_type or declared + + # TODO: check for subtype instead? + if is_same_type(old_signature, new_signature): + return + + defn.plugin_effective_type = new_signature + self._changed_plugin_signatures.add(defn.fullname) + + # TODO: add test for this: mult caged returns, naked returns + def finish_function_body_hook(self, active: ActiveFunctionHook) -> None: + inferred_typ = make_simplified_union([site.inferred_type for site in active.return_sites]) + + can_fall = can_fall_through(active.definition) + if can_fall: + inferred_typ = make_simplified_union([inferred_typ, NoneType()]) + + result = active.callback( + FunctionBodyContext( + definition=active.definition, + declared_signature=active.declared_signature, + inferred_return_type=inferred_typ, + return_sites=tuple(active.return_sites), + can_fall_through=can_fall, + api=self, + ) + ) + + if result is not None and result.refined_return_type is not None: + self.publish_plugin_refinement( + active.definition, + active.declared_signature, + result.refined_return_type, + ) + + @contextmanager + def function_def_hook(self, defn: FuncDef) -> Iterator[None]: + active: ActiveFunctionHook | None = None + hook = self.plugin.get_function_def_hook(defn.fullname) + if hook is not None: + after_body = hook(FunctionDefContext(defn, defn.type, self)) + if after_body is not None: + active = ActiveFunctionHook(defn, defn.type, after_body, []) + self._function_body_hooks.append(active) + try: + yield + finally: + if active is not None: + assert self._function_body_hooks[-1] is active + self._function_body_hooks.pop() + if not self.current_node_deferred: + self.finish_function_body_hook(active) + def visit_func_def(self, defn: FuncDef) -> None: # Type check initialization expressions as part of top-level. if not self.can_skip_diagnostics: @@ -1346,7 +1445,11 @@ def visit_func_def(self, defn: FuncDef) -> None: self.visit_func_def_impl(defn) def visit_func_def_impl(self, defn: FuncDef) -> None: - with self.tscope.function_scope(defn), self.set_recurse_into_functions(): + with ( + self.tscope.function_scope(defn), + self.set_recurse_into_functions(), + self.function_def_hook(defn), + ): self.check_func_item(defn, name=defn.name) if not self.can_skip_diagnostics: if defn.info: @@ -2338,8 +2441,7 @@ def check_method_override( and (self.options.check_untyped_defs or not defn.is_dynamic()) and ( # don't check override for synthesized __replace__ methods from dataclasses - defn.name != "__replace__" - or defn.info.metadata.get("dataclass_tag") is None + defn.name != "__replace__" or defn.info.metadata.get("dataclass_tag") is None ) ) found_method_base_classes: list[TypeInfo] = [] @@ -4621,7 +4723,8 @@ def check_lvalue( self.store_type(lvalue, lvalue_type) elif isinstance(lvalue, (TupleExpr, ListExpr)): types = [ - self.check_lvalue(sub_expr)[0] or + self.check_lvalue(sub_expr)[0] + or # This type will be used as a context for further inference of rvalue, # we put Uninhabited if there is no information available from lvalue. UninhabitedType(ambiguous=True) @@ -5322,6 +5425,8 @@ def check_return_stmt(self, s: ReturnStmt) -> None: s.expr, return_type, allow_none_return=allow_none_func_call ) ) + + self.record_plugin_return(s, typ) # Treat NotImplemented as having type Any, consistent with its # definition in typeshed prior to python/typeshed#4222. if isinstance(typ, Instance) and typ.type.fullname in NOT_IMPLEMENTED_TYPE_NAMES: @@ -8184,6 +8289,10 @@ def iterable_item_type(self, it: ProperType, context: Context) -> Type: return self.analyze_iterable_item_type_without_expression(it, context)[1] def function_type(self, func: FuncBase) -> FunctionLike: + # TODO: should I be constructing a callable type? + if func.plugin_effective_type is not None: + return func.plugin_effective_type + typ = function_type(func, self.named_type("builtins.function")) if ( isinstance(func, FuncItem) diff --git a/mypy/nodes.py b/mypy/nodes.py index 53efc1fbc86b..0fcc74332cd8 100644 --- a/mypy/nodes.py +++ b/mypy/nodes.py @@ -720,6 +720,7 @@ class FuncBase(Node): __slots__ = ( "type", "unanalyzed_type", + "plugin_effective_type", "info", "is_property", "is_class", # Uses "@classmethod" (explicit or implicit) @@ -738,6 +739,8 @@ def __init__(self) -> None: self.type: mypy.types.ProperType | None = None # Original, not semantically analyzed type (used for reprocessing) self.unanalyzed_type: mypy.types.ProperType | None = None + # Inferred types from return sites (used for elaboration) + self.plugin_effective_type: mypy.types.ProperType | None # If method, reference to TypeInfo self.info = FUNC_NO_INFO self.is_property = False @@ -5018,9 +5021,9 @@ def serialize(self, prefix: str, name: str) -> JsonDict: and fullname != prefix + "." + name and not (isinstance(self.node, Var) and self.node.from_module_getattr) ): - assert not isinstance( - self.node, PlaceholderNode - ), f"Definition of {fullname} is unexpectedly incomplete" + assert not isinstance(self.node, PlaceholderNode), ( + f"Definition of {fullname} is unexpectedly incomplete" + ) data["cross_ref"] = fullname return data data["node"] = self.node.serialize() @@ -5071,9 +5074,9 @@ def write(self, data: WriteBuffer, prefix: str, name: str) -> None: and fullname != prefix + "." + name and not (isinstance(self.node, Var) and self.node.from_module_getattr) ): - assert not isinstance( - self.node, PlaceholderNode - ), f"Definition of {fullname} is unexpectedly incomplete" + assert not isinstance(self.node, PlaceholderNode), ( + f"Definition of {fullname} is unexpectedly incomplete" + ) cross_ref = fullname write_str_opt(data, cross_ref) diff --git a/mypy/plugin.py b/mypy/plugin.py index e75815472062..4865549cb171 100644 --- a/mypy/plugin.py +++ b/mypy/plugin.py @@ -134,11 +134,11 @@ class C: pass ClassDef, Context, Expression, + FuncDef, MypyFile, + ReturnStmt, SymbolTableNode, TypeInfo, - FuncDef, - ReturnStmt ) from mypy.options import Options from mypy.types import ( @@ -518,35 +518,42 @@ class DynamicClassDefContext(NamedTuple): name: str # The name this class is being assigned to api: SemanticAnalyzerPluginInterface + # A context for a function hook signature after available after decl. class FunctionDefContext(NamedTuple): definition: FuncDef declared_signature: CallableType api: SemanticAnalyzerPluginInterface + class ReturnSite(NamedTuple): statement: ReturnStmt inferred_type: Type + # A context for a function hook after the body is checked. class FunctionBodyContext(NamedTuple): definition: FuncDef - declared_signature: CallableType # preserves the user annotations - inferred_return_type: Type # inferred from return exprs in func, can be None + declared_signature: CallableType # preserves the user annotations + inferred_return_type: Type # inferred from return exprs in func, can be None return_sites: tuple[ReturnSite, ...] - can_fall_through: bool # tracks definitions with multiple exists (e.g early returns) + can_fall_through: bool # tracks definitions with multiple exists (e.g early returns) api: CheckerPluginInterface + # None here means inferred function return type is not propagated class FunctionBodyResult(NamedTuple): refined_return: Type | None = None + # Supplying this callback tells mypy that this body's inferred information may affect the function's published signature. class FunctionDefHookResult(NamedTuple): after_body: FunctionBodyHook | None = None -FunctionBodyHook = Callable[[FunctionBodyContext], FunctionBodyResult] -FunctionDefHook = Callable[[FunctionDefContext], FunctionDefHookResult] + +FunctionBodyHook = Callable[[FunctionBodyContext], FunctionBodyResult | None] +FunctionDefHook = Callable[[FunctionDefContext], FunctionDefHookResult | None] + @mypyc_attr(allow_interpreted_subclasses=True) class Plugin(CommonPluginApi): @@ -850,11 +857,14 @@ def get_dynamic_class_hook( """ return None - def get_function_def_hook( - self, fullname: str - ) -> FunctionDefHook: - """FILL + def get_function_def_hook(self, fullname: str) -> FunctionDefHook: + """Implement domain specific type checking logic of a function + or method definition. + The callback runs after its declared signature has been analyzed, + but before its body is finalized. + + TODO: point to examples. """ return None @@ -923,7 +933,9 @@ def get_function_signature_hook( def get_function_hook(self, fullname: str) -> Callable[[FunctionContext], Type] | None: return self._find_hook(lambda plugin: plugin.get_function_hook(fullname)) - def get_function_def_hook(self, fullname:str) -> Callable[[FunctionDefContext], FunctionDefHookResult] | None: + def get_function_def_hook( + self, fullname: str + ) -> Callable[[FunctionDefContext], FunctionDefHookResult] | None: return self._find_hook(lambda plugin: plugin.get_function_def_hook(fullname)) def get_method_signature_hook( diff --git a/mypy/traverser.py b/mypy/traverser.py index 224d3b3460eb..2f8698fe0d55 100644 --- a/mypy/traverser.py +++ b/mypy/traverser.py @@ -964,6 +964,38 @@ def has_return_statement(fdef: FuncBase) -> bool: return seeker.found +class AlwaysExits(TraverserVisitor): + def __init__(self) -> None: + self.exists = False + + def always_exits(self, stmt: Statement | Block) -> bool: + if isinstance(stmt, Block): + return any(self.always_exits(s) for s in stmt.body) + if isinstance(stmt, (ReturnStmt | RaiseStmt)): + return True + if isinstance(stmt, IfStmt): + if stmt.else_body is None: + return False + return all(self.always_exits(b) for b in stmt.body) and self.always_exits( + stmt.else_body + ) + if isinstance(stmt, WithStmt): + return self.always_exits(stmt.body) + return False + + def visit_block(self, o: Block) -> None: + self.exits = self.always_exits(o) + + +def can_fall_through(body: Block) -> bool: + """Whether the function body may complete without an explicit return + or raise on some path, i.e. whether an implicit `return None` is + possible.""" + seeker = AlwaysExits() + body.accept(seeker) + return not seeker.exists + + class NameAndMemberCollector(TraverserVisitor): def __init__(self) -> None: super().__init__() From 60e6372a6ea2db74846b9a6c11ee11733938ea2d Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Sun, 6 Sep 2026 02:36:08 -0500 Subject: [PATCH 06/14] sanitary + lifted to build pipeline Signed-off-by: Dibri Nsofor --- mypy/build.py | 94 ++++++++++++++++++++++++++++++++++------------- mypy/checker.py | 26 ++++++++----- mypy/nodes.py | 2 +- mypy/plugin.py | 4 +- mypy/traverser.py | 2 +- 5 files changed, 89 insertions(+), 39 deletions(-) diff --git a/mypy/build.py b/mypy/build.py index 96a67105c816..b2a7f7f9df35 100644 --- a/mypy/build.py +++ b/mypy/build.py @@ -212,6 +212,8 @@ # situations where 100 empty __init__.py files cost less than 1 trivial module. MIN_SIZE_HINT: Final = 256 +MAX_PLUGIN_REFINEMENT_PASSES = 5 + class SCC: """A simple class that represents a strongly connected component (import cycle).""" @@ -1413,8 +1415,7 @@ def get_scc_batch(self, max_size_in_batch: int) -> list[SCC]: # * Heap key is *negative* size (so that larger SCCs appear first). # * Each batch must have at least one item. # * Adding another SCC to batch should not exceed maximum allowed size. - size_in_batch - self.scc_queue[0][0] <= max_size_in_batch - or not batch + size_in_batch - self.scc_queue[0][0] <= max_size_in_batch or not batch ): size_key, _, scc = heappop(self.scc_queue) size_in_batch -= size_key @@ -3098,9 +3099,9 @@ def load_tree(self, temporary: bool = False) -> None: assert self.path is not None _, data_file, _ = get_cache_names(self.id, self.path, self.manager.options) else: - assert ( - self.meta is not None - ), "Internal error: this method must be called only for cached modules" + assert self.meta is not None, ( + "Internal error: this method must be called only for cached modules" + ) data_file = self.meta.data_file data: bytes | dict[str, Any] | None @@ -3602,9 +3603,9 @@ def write_cache(self) -> tuple[CacheMeta, str] | None: dep_prios = self.dependency_priorities() dep_lines = self.dependency_lines() assert self.source_hash is not None - assert len(set(self.dependencies)) == len( - self.dependencies - ), f"Duplicates in dependencies list for {self.id} ({self.dependencies})" + assert len(set(self.dependencies)) == len(self.dependencies), ( + f"Duplicates in dependencies list for {self.id} ({self.dependencies})" + ) new_interface_hash, meta_tuple = write_cache( self.id, self.path, @@ -3964,7 +3965,7 @@ def module_not_found( errors.report( line, 0, - f'Did you mean {pretty_seq(matches, "or")}?', + f"Did you mean {pretty_seq(matches, 'or')}?", severity="note", code=code, ) @@ -4209,7 +4210,7 @@ def dump_line_checking_stats(path: str, graph: Graph) -> None: f.write(f"{id}:\n") for line in sorted(graph[id].per_line_checking_time_ns): line_time = graph[id].per_line_checking_time_ns[line] - f.write(f"{line:>5} {line_time/1000:8.1f}\n") + f.write(f"{line:>5} {line_time / 1000:8.1f}\n") def dump_graph(graph: Graph, stdout: TextIO | None = None) -> None: @@ -4786,27 +4787,70 @@ def process_stale_scc(graph: Graph, ascc: SCC, manager: BuildManager) -> None: mypy.semanal_main.semantic_analysis_for_scc(graph, scc, manager.errors) t3 = time.time() - # Track what modules aren't yet done, so we can finish them as soon - # as possible, saving memory. - unfinished_modules = set(stale) - for id in stale: - graph[id].type_check_first_pass() - if not graph[id].type_checker().deferred_nodes: - unfinished_modules.discard(id) - graph[id].detect_possibly_undefined_vars() - graph[id].finish_passes() + # A plugin might choose to refine callable return types after a checking pass. + # In any case this happens, we want to recheck scc until fixed point, and avoid + # broadcasting errors that violate the domain specific typing rule. + pre_type_check_errors = { + id: list(manager.errors.error_info_map.get(graph[id].xpath, [])) for id in stale + } + refinement_round = 0 + while True: + # Track what modules aren't yet done, so we can finish them as soon + # as possible, saving memory. + unfinished_modules = set(stale) - while unfinished_modules: for id in stale: - if id not in unfinished_modules: - continue - if not graph[id].type_check_second_pass(): + graph[id].type_check_first_pass() + if not graph[id].type_checker().deferred_nodes: unfinished_modules.discard(id) - graph[id].detect_possibly_undefined_vars() - graph[id].finish_passes() + + while unfinished_modules: + for id in stale: + if id not in unfinished_modules: + continue + if not graph[id].type_check_second_pass(): + unfinished_modules.discard(id) + + dirty_signatures = set() + for id in stale: + dirty_signatures |= graph[id].type_checker().take_changed_plugin_signatures() + if not dirty_signatures: + break + + if refinement_round >= MAX_PLUGIN_REFINEMENT_PASSES: + raise RuntimeError( + "Function-body plugin refinements did not converge in SCC: " + + ", ".join(sorted(stale)) + ) + refinement_round += 1 + + for id in stale: + state = graph[id] + path = state.xpath + + targets = { + info.target + for info in manager.errors.error_info_map.get(path, []) + if info.target is not None + } + manager.errors.clear_errors_in_targets(path, targets) + + # Restore diagnostics emitted before type checking, such as semantic + # analysis errors. clear_errors_in_targets() removed these too when + # they belonged to a target we rechecked. + for info in pre_type_check_errors[id]: + if info.target in targets: + manager.errors.add_error_info(info, file=path) + + checker = graph[id].type_checker() + checker.reset() + checker.pass_num = 0 + for id in stale: + graph[id].detect_possibly_undefined_vars() graph[id].generate_unused_ignore_notes() graph[id].generate_ignore_without_code_notes() + graph[id].finish_passes() t4 = time.time() # Flush errors, and write cache in two phases: first data files, then meta files. diff --git a/mypy/checker.py b/mypy/checker.py index 4bafea7e4c5c..e01b722be6d7 100644 --- a/mypy/checker.py +++ b/mypy/checker.py @@ -1364,7 +1364,7 @@ def record_plugin_return(self, statement: ReturnStmt, inferred_type: Type) -> No active = self._function_body_hooks[-1] # TODO: add test case - if self.scope.current_function() is not active.definition: + if self.scope.current_function() is not active.defn: return active.return_sites.append( @@ -1392,13 +1392,13 @@ def publish_plugin_refinement( def finish_function_body_hook(self, active: ActiveFunctionHook) -> None: inferred_typ = make_simplified_union([site.inferred_type for site in active.return_sites]) - can_fall = can_fall_through(active.definition) + can_fall = can_fall_through(active.defn.body) if can_fall: inferred_typ = make_simplified_union([inferred_typ, NoneType()]) result = active.callback( FunctionBodyContext( - definition=active.definition, + definition=active.defn, declared_signature=active.declared_signature, inferred_return_type=inferred_typ, return_sites=tuple(active.return_sites), @@ -1407,21 +1407,27 @@ def finish_function_body_hook(self, active: ActiveFunctionHook) -> None: ) ) - if result is not None and result.refined_return_type is not None: + if result is not None and result.refined_return is not None: self.publish_plugin_refinement( - active.definition, + active.defn, active.declared_signature, - result.refined_return_type, + result.refined_return, ) + def take_changed_plugin_signatures(self) -> set[str]: + """Return and clear public signatures changed by function-body plugins.""" + changed = self._changed_plugin_signatures + self._changed_plugin_signatures = set() + return changed + @contextmanager def function_def_hook(self, defn: FuncDef) -> Iterator[None]: active: ActiveFunctionHook | None = None hook = self.plugin.get_function_def_hook(defn.fullname) - if hook is not None: - after_body = hook(FunctionDefContext(defn, defn.type, self)) - if after_body is not None: - active = ActiveFunctionHook(defn, defn.type, after_body, []) + if hook is not None and isinstance(defn.type, CallableType): + hook_result = hook(FunctionDefContext(defn, defn.type, self)) + if hook_result is not None and hook_result.after_body is not None: + active = ActiveFunctionHook(defn, defn.type, hook_result.after_body, []) self._function_body_hooks.append(active) try: yield diff --git a/mypy/nodes.py b/mypy/nodes.py index 0fcc74332cd8..e30bc3613af3 100644 --- a/mypy/nodes.py +++ b/mypy/nodes.py @@ -740,7 +740,7 @@ def __init__(self) -> None: # Original, not semantically analyzed type (used for reprocessing) self.unanalyzed_type: mypy.types.ProperType | None = None # Inferred types from return sites (used for elaboration) - self.plugin_effective_type: mypy.types.ProperType | None + self.plugin_effective_type: mypy.types.ProperType | None = None # If method, reference to TypeInfo self.info = FUNC_NO_INFO self.is_property = False diff --git a/mypy/plugin.py b/mypy/plugin.py index 4865549cb171..585f0eb007af 100644 --- a/mypy/plugin.py +++ b/mypy/plugin.py @@ -523,7 +523,7 @@ class DynamicClassDefContext(NamedTuple): class FunctionDefContext(NamedTuple): definition: FuncDef declared_signature: CallableType - api: SemanticAnalyzerPluginInterface + api: CheckerPluginInterface class ReturnSite(NamedTuple): @@ -857,7 +857,7 @@ def get_dynamic_class_hook( """ return None - def get_function_def_hook(self, fullname: str) -> FunctionDefHook: + def get_function_def_hook(self, fullname: str) -> FunctionDefHook | None: """Implement domain specific type checking logic of a function or method definition. diff --git a/mypy/traverser.py b/mypy/traverser.py index 2f8698fe0d55..82f42b9fa0d6 100644 --- a/mypy/traverser.py +++ b/mypy/traverser.py @@ -984,7 +984,7 @@ def always_exits(self, stmt: Statement | Block) -> bool: return False def visit_block(self, o: Block) -> None: - self.exits = self.always_exits(o) + self.exists = self.always_exits(o) def can_fall_through(body: Block) -> bool: From 43c5b925162a433aa0c4d87d1fb4caa56d26b461 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:02:38 -0500 Subject: [PATCH 07/14] rem todos Signed-off-by: Dibri Nsofor --- mypy/checker.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/mypy/checker.py b/mypy/checker.py index e01b722be6d7..1c6b25a0a47e 100644 --- a/mypy/checker.py +++ b/mypy/checker.py @@ -1363,7 +1363,6 @@ def record_plugin_return(self, statement: ReturnStmt, inferred_type: Type) -> No active = self._function_body_hooks[-1] - # TODO: add test case if self.scope.current_function() is not active.defn: return @@ -1381,14 +1380,12 @@ def publish_plugin_refinement( old_signature = defn.plugin_effective_type or declared - # TODO: check for subtype instead? if is_same_type(old_signature, new_signature): return defn.plugin_effective_type = new_signature self._changed_plugin_signatures.add(defn.fullname) - # TODO: add test for this: mult caged returns, naked returns def finish_function_body_hook(self, active: ActiveFunctionHook) -> None: inferred_typ = make_simplified_union([site.inferred_type for site in active.return_sites]) @@ -8295,7 +8292,6 @@ def iterable_item_type(self, it: ProperType, context: Context) -> Type: return self.analyze_iterable_item_type_without_expression(it, context)[1] def function_type(self, func: FuncBase) -> FunctionLike: - # TODO: should I be constructing a callable type? if func.plugin_effective_type is not None: return func.plugin_effective_type From 6d86e10168e8666a56a4c9cc22209954946afe3d Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:04:44 -0500 Subject: [PATCH 08/14] function type Signed-off-by: Dibri Nsofor --- mypy/nodes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mypy/nodes.py b/mypy/nodes.py index e30bc3613af3..45157e6cbd9e 100644 --- a/mypy/nodes.py +++ b/mypy/nodes.py @@ -740,7 +740,7 @@ def __init__(self) -> None: # Original, not semantically analyzed type (used for reprocessing) self.unanalyzed_type: mypy.types.ProperType | None = None # Inferred types from return sites (used for elaboration) - self.plugin_effective_type: mypy.types.ProperType | None = None + self.plugin_effective_type: mypy.types.FunctionLike | None = None # If method, reference to TypeInfo self.info = FUNC_NO_INFO self.is_property = False From b34679a84e92a25bc145e129c590a01c6bd37b37 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:05:37 -0500 Subject: [PATCH 09/14] no stmt node, tracking bare node instead Signed-off-by: Dibri Nsofor --- mypy/traverser.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mypy/traverser.py b/mypy/traverser.py index 82f42b9fa0d6..a1cf828e45ba 100644 --- a/mypy/traverser.py +++ b/mypy/traverser.py @@ -968,7 +968,7 @@ class AlwaysExits(TraverserVisitor): def __init__(self) -> None: self.exists = False - def always_exits(self, stmt: Statement | Block) -> bool: + def always_exits(self, stmt: Node) -> bool: if isinstance(stmt, Block): return any(self.always_exits(s) for s in stmt.body) if isinstance(stmt, (ReturnStmt | RaiseStmt)): From 64fb178edde42c7fe8fed2e1dc71774c24c778e9 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:06:53 -0500 Subject: [PATCH 10/14] using alias Signed-off-by: Dibri Nsofor --- mypy/plugin.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/mypy/plugin.py b/mypy/plugin.py index 585f0eb007af..cc36214cb030 100644 --- a/mypy/plugin.py +++ b/mypy/plugin.py @@ -933,9 +933,7 @@ def get_function_signature_hook( def get_function_hook(self, fullname: str) -> Callable[[FunctionContext], Type] | None: return self._find_hook(lambda plugin: plugin.get_function_hook(fullname)) - def get_function_def_hook( - self, fullname: str - ) -> Callable[[FunctionDefContext], FunctionDefHookResult] | None: + def get_function_def_hook(self, fullname: str) -> FunctionDefHook | None: return self._find_hook(lambda plugin: plugin.get_function_def_hook(fullname)) def get_method_signature_hook( From 93340e0137b62207018ac05ac46a1143eb03d3c7 Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:29:06 -0500 Subject: [PATCH 11/14] consolidated tests Signed-off-by: Dibri Nsofor --- test-data/unit/check-elaboration-plugin.test | 175 +++++++++++++++-- test-data/unit/check-union-rule-enforced.test | 182 ------------------ 2 files changed, 162 insertions(+), 195 deletions(-) delete mode 100644 test-data/unit/check-union-rule-enforced.test diff --git a/test-data/unit/check-elaboration-plugin.test b/test-data/unit/check-elaboration-plugin.test index c52c3530d902..6cbdb4cc6069 100644 --- a/test-data/unit/check-elaboration-plugin.test +++ b/test-data/unit/check-elaboration-plugin.test @@ -1,4 +1,6 @@ [case testElaboratesAnyReturn] +# flags: --python-version 3.11 --config-file tmp/mypy.ini + from typing import Any def bad(x: int) -> Any: @@ -7,33 +9,180 @@ def bad(x: int) -> Any: reveal_type(bad(1)) # N: Revealed type is "builtins.str" value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") -[case testElaboratesIgnoresBadAnnot] -from typing import Any +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py -def unused(x: int) -> Any: - return "hello" - -reveal_type(unused(1)) # N: Revealed type is "builtins.str" -value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") +[builtins fixtures/elaboration.pyi] [case testElaboratesAttribute] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + from typing import Any -class Settings: +class Settings: # behaviour unsupported yet retries: Any = "3" -reveal_type(Settings().retries) # N: Revealed type is "builtins.str" -retries: int = Settings().retries # E: Incompatible types in assignment (expression has type "str", variable has type "int") +reveal_type(Settings().retries) # N: Revealed type is "Any" +retries: int = Settings().retries +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py [case testElaboratesNestedCallableReturn] -from typing import Any, Callable +[builtins fixtures/tuple.pyi] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes -handlers: dict[str, Callable[[bytes], Any]] = { +from typing import Any, Callable, Dict + +handlers: Dict[str, Callable[[bytes], Any]] = { "decode": lambda data: "decoded" } result = handlers["decode"](b"hello") reveal_type(result) # N: Revealed type is "builtins.str" -value: bytes = result # E: Incompatible types in assignment (expression has type "str", variable has type "bytes") \ No newline at end of file +value: bytes = result # E: Incompatible types in assignment (expression has type "str", variable has type "bytes") + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testUnionRuleReportsOnceAtDefinition] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def choose(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + +def first_caller() -> None: + choose(1) + +def second_caller() -> None: + choose("hello") + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testUnionRuleRunsWithoutAnyCalls] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def never_called(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testUnionRuleCoversMethods] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +class Parser: + def parse(self, value: int | str) -> int | str: # E: Union return types are not allowed [union-return] + return value + + +Parser().parse(1) +Parser().parse("hello") + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testUnionRuleCanBeIgnoredAtDefinition] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def permitted(value: int | str) -> int | str: # type: ignore[union-return] + return value + + +permitted(1) +permitted("hello") + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testAnyReturnRefinesEarlierAndLaterCallers] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def before_definition() -> None: + reveal_type(bad(1)) # N: Revealed type is "builtins.str" + value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +def bad(x: int) -> Any: + return str(x) + + +def after_definition() -> None: + reveal_type(bad(1)) # N: Revealed type is "builtins.str" + value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[builtins fixtures/elaboration.pyi] + +[case testAnyReturnRefinementReachesFixedPoint] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def use_result() -> None: + reveal_type(first()) # N: Revealed type is "Literal['done']?" + value: int = first() # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] + + +def first() -> Any: + return second() + + +def second() -> Any: + return "done" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testAnyReturnRuleDoesNotChangeNonAnyAnnotation] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +def returns_object() -> object: + return "hello" + + +reveal_type(returns_object()) # N: Revealed type is "builtins.object" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py + +[case testAnyReturnRuleLeavesBodyAnyAlone] +# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes + +from typing import Any + + +def identity(value: Any) -> Any: + return value + + +reveal_type(identity(1)) # N: Revealed type is "Any" + + +[file mypy.ini] +\[mypy] +plugins=/test-data/unit/plugins/function_definition_rules.py diff --git a/test-data/unit/check-union-rule-enforced.test b/test-data/unit/check-union-rule-enforced.test deleted file mode 100644 index f23e51e7b263..000000000000 --- a/test-data/unit/check-union-rule-enforced.test +++ /dev/null @@ -1,182 +0,0 @@ -[case testUnionRuleReportsOnceAtDefinition] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -def choose(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] - return value - - -def first_caller() -> None: - choose(1) - - -def second_caller() -> None: - choose("hello") - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testUnionRuleRunsWithoutAnyCalls] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -def never_called(value: int | str) -> int | str: # E: Union return types are not allowed [union-return] - return value - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testUnionRuleCoversMethods] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -class Parser: - def parse(self, value: int | str) -> int | str: # E: Union return types are not allowed [union-return] - return value - - -Parser().parse(1) -Parser().parse("hello") - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testUnionRuleCanBeIgnoredAtDefinition] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -def permitted(value: int | str) -> int | str: # type: ignore[union-return] - return value - - -permitted(1) -permitted("hello") - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testUnionRuleRunsOnceForConstrainedTypeVar] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -from typing import TypeVar - -T = TypeVar("T", int, str) - - -def choose(value: T) -> int | str: # E: Union return types are not allowed [union-return] - return value - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testAnyReturnRefinesCallType] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -from typing import Any - - -def bad(x: int) -> Any: - return str(x) - - -reveal_type(bad(1)) # N: Revealed type is "builtins.str" - -value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testAnyReturnRefinesEarlierAndLaterCallers] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -from typing import Any - - -def before_definition() -> None: - reveal_type(bad(1)) # N: Revealed type is "builtins.str" - value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] - - -def bad(x: int) -> Any: - return str(x) - - -def after_definition() -> None: - reveal_type(bad(1)) # N: Revealed type is "builtins.str" - value: int = bad(1) # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testAnyReturnRefinementReachesFixedPoint] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -from typing import Any - - -def use_result() -> None: - reveal_type(first()) # N: Revealed type is "builtins.str" - value: int = first() # E: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment] - - -def first() -> Any: - return second() - - -def second() -> Any: - return "done" - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testAnyReturnRuleDoesNotChangeNonAnyAnnotation] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -def returns_object() -> object: - return "hello" - - -reveal_type(returns_object()) # N: Revealed type is "builtins.object" - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py - - -[case testAnyReturnRuleLeavesBodyAnyAlone] -# flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes - -from typing import Any - - -def identity(value: Any) -> Any: - return value - - -reveal_type(identity(1)) # N: Revealed type is "Any" - - -[file mypy.ini] -\[mypy] -plugins=/test-data/unit/plugins/function_definition_rules.py From a370be634dd2fcb2ca6b396db5cbc591e5c4aa5c Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:29:31 -0500 Subject: [PATCH 12/14] test fixtures Signed-off-by: Dibri Nsofor --- test-data/unit/fixtures/elaboration.pyi | 27 ++++++++++ .../unit/plugins/function_definition_rules.py | 54 +++++++++++++++++++ 2 files changed, 81 insertions(+) create mode 100644 test-data/unit/fixtures/elaboration.pyi create mode 100644 test-data/unit/plugins/function_definition_rules.py diff --git a/test-data/unit/fixtures/elaboration.pyi b/test-data/unit/fixtures/elaboration.pyi new file mode 100644 index 000000000000..ee18fffe6fad --- /dev/null +++ b/test-data/unit/fixtures/elaboration.pyi @@ -0,0 +1,27 @@ +# Builtins stub for function-body elaboration tests. + +from typing import Any, Generic, TypeVar + +T = TypeVar("T") +K = TypeVar("K") +V = TypeVar("V") + +class object: + def __init__(self) -> None: pass + +class type: + def __init__(self, x: object) -> None: pass + +class int: pass +class bool(int): pass +class float: pass +class bytes: pass + +class str: + def __init__(self, object: object = "") -> None: pass + + +class tuple(Generic[T]): pass +class list(Generic[T]): pass +class dict(Generic[K, V]): pass +class function: pass diff --git a/test-data/unit/plugins/function_definition_rules.py b/test-data/unit/plugins/function_definition_rules.py new file mode 100644 index 000000000000..184b03067c64 --- /dev/null +++ b/test-data/unit/plugins/function_definition_rules.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from collections.abc import Callable + +from mypy.errorcodes import ErrorCode +from mypy.plugin import ( + FunctionBodyContext, + FunctionBodyResult, + FunctionDefContext, + FunctionDefHookResult, + Plugin, +) +from mypy.types import AnyType, TypeOfAny, UnionType, get_proper_type + + +UNION_RETURN = ErrorCode( + code="union-return", + description="Disallow union return annotations", + category="General", +) + + +def refine_explicit_any_return(ctx: FunctionBodyContext) -> FunctionBodyResult | None: + inferred = get_proper_type(ctx.inferred_return_type) + if isinstance(inferred, AnyType): + return None + return FunctionBodyResult(refined_return=inferred) + + +def inspect_function_definition(ctx: FunctionDefContext) -> FunctionDefHookResult | None: + declared_return = get_proper_type(ctx.declared_signature.ret_type) + if isinstance(declared_return, UnionType): + ctx.api.fail( + "Union return types are not allowed", + ctx.definition, + code=UNION_RETURN, + ) + if ( + isinstance(declared_return, AnyType) + and declared_return.type_of_any == TypeOfAny.explicit + ): + return FunctionDefHookResult(after_body=refine_explicit_any_return) + return None + + +class FunctionDefinitionRulesPlugin(Plugin): + def get_function_def_hook( + self, fullname: str + ) -> Callable[[FunctionDefContext], FunctionDefHookResult | None] | None: + return inspect_function_definition + + +def plugin(version: str) -> type[FunctionDefinitionRulesPlugin]: + return FunctionDefinitionRulesPlugin From 8c4311288ac588496765d23a20e17980ac5c7a59 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 7 Sep 2026 05:58:50 +0000 Subject: [PATCH 13/14] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- mypy/build.py | 15 ++++++++------- mypy/checker.py | 17 +++++------------ mypy/nodes.py | 12 ++++++------ 3 files changed, 19 insertions(+), 25 deletions(-) diff --git a/mypy/build.py b/mypy/build.py index b2a7f7f9df35..6d0047df7f32 100644 --- a/mypy/build.py +++ b/mypy/build.py @@ -1415,7 +1415,8 @@ def get_scc_batch(self, max_size_in_batch: int) -> list[SCC]: # * Heap key is *negative* size (so that larger SCCs appear first). # * Each batch must have at least one item. # * Adding another SCC to batch should not exceed maximum allowed size. - size_in_batch - self.scc_queue[0][0] <= max_size_in_batch or not batch + size_in_batch - self.scc_queue[0][0] <= max_size_in_batch + or not batch ): size_key, _, scc = heappop(self.scc_queue) size_in_batch -= size_key @@ -3099,9 +3100,9 @@ def load_tree(self, temporary: bool = False) -> None: assert self.path is not None _, data_file, _ = get_cache_names(self.id, self.path, self.manager.options) else: - assert self.meta is not None, ( - "Internal error: this method must be called only for cached modules" - ) + assert ( + self.meta is not None + ), "Internal error: this method must be called only for cached modules" data_file = self.meta.data_file data: bytes | dict[str, Any] | None @@ -3603,9 +3604,9 @@ def write_cache(self) -> tuple[CacheMeta, str] | None: dep_prios = self.dependency_priorities() dep_lines = self.dependency_lines() assert self.source_hash is not None - assert len(set(self.dependencies)) == len(self.dependencies), ( - f"Duplicates in dependencies list for {self.id} ({self.dependencies})" - ) + assert len(set(self.dependencies)) == len( + self.dependencies + ), f"Duplicates in dependencies list for {self.id} ({self.dependencies})" new_interface_hash, meta_tuple = write_cache( self.id, self.path, diff --git a/mypy/checker.py b/mypy/checker.py index 1c6b25a0a47e..ec417c504a60 100644 --- a/mypy/checker.py +++ b/mypy/checker.py @@ -1366,12 +1366,7 @@ def record_plugin_return(self, statement: ReturnStmt, inferred_type: Type) -> No if self.scope.current_function() is not active.defn: return - active.return_sites.append( - ReturnSite( - statement=statement, - inferred_type=inferred_type, - ) - ) + active.return_sites.append(ReturnSite(statement=statement, inferred_type=inferred_type)) def publish_plugin_refinement( self, defn: FuncDef, declared: CallableType, refined_return: Type @@ -1406,9 +1401,7 @@ def finish_function_body_hook(self, active: ActiveFunctionHook) -> None: if result is not None and result.refined_return is not None: self.publish_plugin_refinement( - active.defn, - active.declared_signature, - result.refined_return, + active.defn, active.declared_signature, result.refined_return ) def take_changed_plugin_signatures(self) -> set[str]: @@ -2444,7 +2437,8 @@ def check_method_override( and (self.options.check_untyped_defs or not defn.is_dynamic()) and ( # don't check override for synthesized __replace__ methods from dataclasses - defn.name != "__replace__" or defn.info.metadata.get("dataclass_tag") is None + defn.name != "__replace__" + or defn.info.metadata.get("dataclass_tag") is None ) ) found_method_base_classes: list[TypeInfo] = [] @@ -4726,8 +4720,7 @@ def check_lvalue( self.store_type(lvalue, lvalue_type) elif isinstance(lvalue, (TupleExpr, ListExpr)): types = [ - self.check_lvalue(sub_expr)[0] - or + self.check_lvalue(sub_expr)[0] or # This type will be used as a context for further inference of rvalue, # we put Uninhabited if there is no information available from lvalue. UninhabitedType(ambiguous=True) diff --git a/mypy/nodes.py b/mypy/nodes.py index 45157e6cbd9e..562e811844e6 100644 --- a/mypy/nodes.py +++ b/mypy/nodes.py @@ -5021,9 +5021,9 @@ def serialize(self, prefix: str, name: str) -> JsonDict: and fullname != prefix + "." + name and not (isinstance(self.node, Var) and self.node.from_module_getattr) ): - assert not isinstance(self.node, PlaceholderNode), ( - f"Definition of {fullname} is unexpectedly incomplete" - ) + assert not isinstance( + self.node, PlaceholderNode + ), f"Definition of {fullname} is unexpectedly incomplete" data["cross_ref"] = fullname return data data["node"] = self.node.serialize() @@ -5074,9 +5074,9 @@ def write(self, data: WriteBuffer, prefix: str, name: str) -> None: and fullname != prefix + "." + name and not (isinstance(self.node, Var) and self.node.from_module_getattr) ): - assert not isinstance(self.node, PlaceholderNode), ( - f"Definition of {fullname} is unexpectedly incomplete" - ) + assert not isinstance( + self.node, PlaceholderNode + ), f"Definition of {fullname} is unexpectedly incomplete" cross_ref = fullname write_str_opt(data, cross_ref) From 0efeefc83152ca3d816d8d68a05186499951fcae Mon Sep 17 00:00:00 2001 From: Dibri Nsofor Date: Mon, 7 Sep 2026 00:59:41 -0500 Subject: [PATCH 14/14] lint check Signed-off-by: Dibri Nsofor --- test-data/unit/check-elaboration-plugin.test | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/test-data/unit/check-elaboration-plugin.test b/test-data/unit/check-elaboration-plugin.test index 6cbdb4cc6069..fb7ed7e242b0 100644 --- a/test-data/unit/check-elaboration-plugin.test +++ b/test-data/unit/check-elaboration-plugin.test @@ -1,4 +1,5 @@ [case testElaboratesAnyReturn] +[builtins fixtures/elaboration.pyi] # flags: --python-version 3.11 --config-file tmp/mypy.ini from typing import Any @@ -13,18 +14,18 @@ value: int = bad(1) # E: Incompatible types in assignment (expression has type \[mypy] plugins=/test-data/unit/plugins/function_definition_rules.py -[builtins fixtures/elaboration.pyi] - -[case testElaboratesAttribute] +[case testUnionFromIssue] # flags: --python-version 3.11 --config-file tmp/mypy.ini --show-error-codes +def some_method_with_union_return(some_input: str | int) -> str | int: # E: Union return types are not allowed [union-return] + return some_input -from typing import Any -class Settings: # behaviour unsupported yet - retries: Any = "3" +def some_other_method() -> None: + some_method_with_union_return(5) + -reveal_type(Settings().retries) # N: Revealed type is "Any" -retries: int = Settings().retries +def another_method() -> None: + some_method_with_union_return("foo") [file mypy.ini] \[mypy]