b623189560
## Summary
Closes: https://github.com/astral-sh/ty/issues/157
This PR adds support for the following capabilities involving a
`ParamSpec` type variable:
- Representing `P.args` and `P.kwargs` in the type system
- Matching against a callable containing `P` to create a type mapping
- Specializing `P` against the stored parameters
The value of a `ParamSpec` type variable is being represented using
`CallableType` with a `CallableTypeKind::ParamSpecValue` variant. This
`CallableTypeKind` is expanded from the existing `is_function_like`
boolean flag. An `enum` is used as these variants are mutually
exclusive.
For context, an initial iteration made an attempt to expand the
`Specialization` to use `TypeOrParameters` enum that represents that a
type variable can specialize into either a `Type` or `Parameters` but
that increased the complexity of the code as all downstream usages would
need to handle both the variants appropriately. Additionally, we'd have
also need to establish an invariant that a regular type variable always
maps to a `Type` while a paramspec type variable always maps to a
`Parameters`.
I've intentionally left out checking and raising diagnostics when the
`ParamSpec` type variable and it's components are not being used
correctly to avoid scope increase and it can easily be done as a
follow-up. This would also include the scoping rules which I don't think
a regular type variable implements either.
## Test Plan
Add new mdtest cases and update existing test cases.
Ran this branch on pyx, no new diagnostics.
### Ecosystem analysis
There's a case where in an annotated assignment like:
```py
type CustomType[P] = Callable[...]
def value[**P](...): ...
def another[**P](...):
target: CustomType[P] = value
```
The type of `value` is a callable and it has a paramspec that's bound to
`value`, `CustomType` is a type alias that's a callable and `P` that's
used in it's specialization is bound to `another`. Now, ty infers the
type of `target` same as `value` and does not use the declared type
`CustomType[P]`. [This is the
assignment](0980b9d9ab/src/async_utils/gen_transform.py (L108))
that I'm referring to which then leads to error in downstream usage.
Pyright and mypy does seem to use the declared type.
There are multiple diagnostics in `dd-trace-py` that requires support
for `cls`.
I'm seeing `Divergent` type for an example like which ~~I'm not sure
why, I'll look into it tomorrow~~ is because of a cycle as mentioned in
https://github.com/astral-sh/ty/issues/1729#issuecomment-3612279974:
```py
from typing import Callable
def decorator[**P](c: Callable[P, int]) -> Callable[P, str]: ...
@decorator
def func(a: int) -> int: ...
# ((a: int) -> str) | ((a: Divergent) -> str)
reveal_type(func)
```
I ~~need to look into why are the parameters not being specialized
through multiple decorators in the following code~~ think this is also
because of the cycle mentioned in
https://github.com/astral-sh/ty/issues/1729#issuecomment-3612279974 and
the fact that we don't support `staticmethod` properly:
```py
from contextlib import contextmanager
class Foo:
@staticmethod
@contextmanager
def method(x: int):
yield
foo = Foo()
# ty: Revealed type: `() -> _GeneratorContextManager[Unknown, None, None]` [revealed-type]
reveal_type(foo.method)
```
There's some issue related to `Protocol` that are generic over a
`ParamSpec` in `starlette` which might be related to
https://github.com/astral-sh/ty/issues/1635 but I'm not sure. Here's a
minimal example to reproduce:
<details><summary>Code snippet:</summary>
<p>
```py
from collections.abc import Awaitable, Callable, MutableMapping
from typing import Any, Callable, ParamSpec, Protocol
P = ParamSpec("P")
Scope = MutableMapping[str, Any]
Message = MutableMapping[str, Any]
Receive = Callable[[], Awaitable[Message]]
Send = Callable[[Message], Awaitable[None]]
ASGIApp = Callable[[Scope, Receive, Send], Awaitable[None]]
_Scope = Any
_Receive = Callable[[], Awaitable[Any]]
_Send = Callable[[Any], Awaitable[None]]
# Since `starlette.types.ASGIApp` type differs from `ASGIApplication` from `asgiref`
# we need to define a more permissive version of ASGIApp that doesn't cause type errors.
_ASGIApp = Callable[[_Scope, _Receive, _Send], Awaitable[None]]
class _MiddlewareFactory(Protocol[P]):
def __call__(
self, app: _ASGIApp, *args: P.args, **kwargs: P.kwargs
) -> _ASGIApp: ...
class Middleware:
def __init__(
self, factory: _MiddlewareFactory[P], *args: P.args, **kwargs: P.kwargs
) -> None:
self.factory = factory
self.args = args
self.kwargs = kwargs
class ServerErrorMiddleware:
def __init__(
self,
app: ASGIApp,
value: int | None = None,
flag: bool = False,
) -> None:
self.app = app
self.value = value
self.flag = flag
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ...
# ty: Argument to bound method `__init__` is incorrect: Expected `_MiddlewareFactory[(...)]`, found `<class 'ServerErrorMiddleware'>` [invalid-argument-type]
Middleware(ServerErrorMiddleware, value=500, flag=True)
```
</p>
</details>
### Conformance analysis
> ```diff
> -constructors_callable.py:36:13: info[revealed-type] Revealed type:
`(...) -> Unknown`
> +constructors_callable.py:36:13: info[revealed-type] Revealed type:
`(x: int) -> Unknown`
> ```
Requires return type inference i.e.,
https://github.com/astral-sh/ruff/pull/21551
> ```diff
> +constructors_callable.py:194:16: error[invalid-argument-type]
Argument is incorrect: Expected `list[T@__init__]`, found `list[Unknown
| str]`
> +constructors_callable.py:194:22: error[invalid-argument-type]
Argument is incorrect: Expected `list[T@__init__]`, found `list[Unknown
| str]`
> +constructors_callable.py:195:4: error[invalid-argument-type] Argument
is incorrect: Expected `list[T@__init__]`, found `list[Unknown | int]`
> +constructors_callable.py:195:9: error[invalid-argument-type] Argument
is incorrect: Expected `list[T@__init__]`, found `list[Unknown | str]`
> ```
I might need to look into why this is happening...
> ```diff
> +generics_defaults.py:79:1: error[type-assertion-failure] Type
`type[Class_ParamSpec[(str, int, /)]]` does not match asserted type
`<class 'Class_ParamSpec'>`
> ```
which is on the following code
```py
DefaultP = ParamSpec("DefaultP", default=[str, int])
class Class_ParamSpec(Generic[DefaultP]): ...
assert_type(Class_ParamSpec, type[Class_ParamSpec[str, int]])
```
It's occurring because there's no equivalence relationship defined
between `ClassLiteral` and `KnownInstanceType::TypeGenericAlias` which
is what these types are.
Everything else looks good to me!
6.7 KiB
6.7 KiB
Async with statements
Basic async with statement
The type of the target variable in a with statement should be the return type from the context
manager's __aenter__ method. However, async with statements aren't supported yet. This test
asserts that it doesn't emit any context manager-related errors.
class Target: ...
class Manager:
async def __aenter__(self) -> Target:
return Target()
async def __aexit__(self, exc_type, exc_value, traceback): ...
async def test():
async with Manager() as f:
reveal_type(f) # revealed: Target
Multiple targets
class Manager:
async def __aenter__(self) -> tuple[int, str]:
return 42, "hello"
async def __aexit__(self, exc_type, exc_value, traceback): ...
async def test():
async with Manager() as (x, y):
reveal_type(x) # revealed: int
reveal_type(y) # revealed: str
Context manager without an __aenter__ or __aexit__ method
class Manager: ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aenter__` and `__aexit__`"
async with Manager():
...
Context manager without an __aenter__ method
class Manager:
async def __aexit__(self, exc_tpe, exc_value, traceback): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aenter__`"
async with Manager():
...
Context manager without an __aexit__ method
class Manager:
async def __aenter__(self): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aexit__`"
async with Manager():
...
Context manager with non-callable __aenter__ attribute
class Manager:
__aenter__: int = 42
async def __aexit__(self, exc_tpe, exc_value, traceback): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not correctly implement `__aenter__`"
async with Manager():
...
Context manager with non-callable __aexit__ attribute
from typing_extensions import Self
class Manager:
def __aenter__(self) -> Self:
return self
__aexit__: int = 32
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not correctly implement `__aexit__`"
async with Manager():
...
Context expression with possibly-unbound union variants
async def _(flag: bool):
class Manager1:
def __aenter__(self) -> str:
return "foo"
def __aexit__(self, exc_type, exc_value, traceback): ...
class NotAContextManager: ...
context_expr = Manager1() if flag else NotAContextManager()
# error: [invalid-context-manager] "Object of type `Manager1 | NotAContextManager` cannot be used with `async with` because the methods `__aenter__` and `__aexit__` are possibly missing"
async with context_expr as f:
reveal_type(f) # revealed: str
Context expression with "sometimes" callable __aenter__ method
async def _(flag: bool):
class Manager:
if flag:
async def __aenter__(self) -> str:
return "abcd"
async def __exit__(self, *args): ...
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because the method `__aenter__` may be missing"
async with Manager() as f:
reveal_type(f) # revealed: CoroutineType[Any, Any, str]
Invalid __aenter__ signature
class Manager:
async def __aenter__() -> str:
return "foo"
async def __aexit__(self, exc_type, exc_value, traceback): ...
async def main():
context_expr = Manager()
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not correctly implement `__aenter__`"
async with context_expr as f:
reveal_type(f) # revealed: CoroutineType[Any, Any, str]
Accidental use of async async with
If a asynchronous async with statement is used on a type with __enter__ and __exit__, we show
a diagnostic hint that the user might have intended to use with instead.
class Manager:
def __enter__(self): ...
def __exit__(self, *args): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aenter__` and `__aexit__`"
async with Manager():
...
Incorrect signatures
The sub-diagnostic is also provided if the signatures of __enter__ and __exit__ do not match the
expected signatures for a context manager:
class Manager:
def __enter__(self): ...
def __exit__(self, typ: str, exc, traceback): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aenter__` and `__aexit__`"
async with Manager():
...
Incorrect number of arguments
Similarly, we also show the hint if the functions have the wrong number of arguments:
class Manager:
def __enter__(self, wrong_extra_arg): ...
def __exit__(self, typ, exc, traceback, wrong_extra_arg): ...
async def main():
# error: [invalid-context-manager] "Object of type `Manager` cannot be used with `async with` because it does not implement `__aenter__` and `__aexit__`"
async with Manager():
...
@asynccontextmanager
from contextlib import asynccontextmanager
from typing import AsyncGenerator
class Session: ...
@asynccontextmanager
async def connect() -> AsyncGenerator[Session]:
yield Session()
# TODO: this should be `() -> _AsyncGeneratorContextManager[Session, None]`
reveal_type(connect) # revealed: () -> _AsyncGeneratorContextManager[Unknown, None]
async def main():
async with connect() as session:
# TODO: should be `Session`
reveal_type(session) # revealed: Unknown
asyncio.timeout
[environment]
python-version = "3.11"
import asyncio
async def long_running_task():
await asyncio.sleep(5)
async def main():
async with asyncio.timeout(1):
await long_running_task()
asyncio.TaskGroup
[environment]
python-version = "3.11"
import asyncio
async def long_running_task():
await asyncio.sleep(5)
async def main():
async with asyncio.TaskGroup() as tg:
reveal_type(tg) # revealed: TaskGroup
tg.create_task(long_running_task())