Removing parameters from the end of a ParamSpec with Concatenate

talley · March 29, 2024, 11:31pm

If this has been addressed elsewhere, please redirect me

From my understanding of PEP612, and the typing docs on generics, typing.Concatenate can be used to denote addition or removal of a parameter from the beginning of a signature. An example (slightly modified from the one in PEP612):

from typing import Callable, Concatenate, ParamSpec, reveal_type, Any

P = ParamSpec("P")

def bar(x: int, y: str) -> int: ...

# higher order func that removes the first parameter
def remove_first(x: Callable[Concatenate[Any, P], int]) -> Callable[P, bool]: ...

reveal_type(remove_first(bar))  #  Revealed type is "def (y: str) -> bool"

I’m wondering why it wasn’t allowed to indicate removal of a paremeter at the end of the spec:

# higher order func that removes the last parameter
# ERR: Last type argument for "Concatenate" must be a ParamSpec or "...
def remove_last(x: Callable[Concatenate[P, Any], int]) -> Callable[P, bool]: ...

reveal_type(remove_last(bar))  # want this to be (x: int) -> bool
# but is actually: Revealed type is "def (*Never, **Never) -> bool

For context, I have a library that implements the observer pattern which will refrain from passing arguments that a callback cannot accept. I would have liked to do the following:

class Signal(Generic[P]):
    # this decorator takes functions with the same Params
    # or one arg removed from the end
    def connect(
        self,
        callback: Callable[P, Any] | Callable[Concatenate[P, Any], Any],
    ) -> Callable: ...

# this signal emits up to 2 arguments, an int and string
signal: Signal[int, str] = Signal()

@signal.connect
def callback1(a: int, y: str): ...  # fine

@signal.connect
def callback1(a: int): ...  # ALSO fine

I suspect the answer has something to do with how this would be achieved. It’s true that I need to do signature inspection to make this work at runtime. But is that always going to be the case here? and is it never the case when removing parameters from the front of the signature? I guess I struggle a bit to see why the scenarios are so different that one was allowed and the other was not.

thanks for your insights!

Jelle · March 29, 2024, 11:46pm

I believe Concatenate was designed the way it was because adding a positional-only parameter at the beginning is just about the only operation that can always be performed on a signature. For example, Concatenate[int, (*args, **kwargs)] = (int, /, *args, **kwargs). (I hope the meaning of this notation is obvious).

How would the following work?

Concatenate[(*args, **kwargs), int] = ?
Concatenate[(*, arg: str), int] = ?

talley · March 29, 2024, 11:59pm

yep, for addition, that absolutely makes sense. No way to add a positional parameter at the end of an arbitrary ParamSpec. Do you agree that for removal though the situation is a bit more complex? or can you see a clear explanation for why that one is also only always safe at the beginning?

edit: perhaps the fact that it cant be used for addition is sufficient, since even if it works for removal, the type checker would need much more context to determine the safety of a Concatentation without a trailing P, or …

Daverball · March 30, 2024, 8:09am

Since, from what I can tell, your API doesn’t actually care about named parameters you can just use TypeVarTuple instead, your example works with that just fine: mypy Playground

from collections.abc import Callable
from typing import Any, Generic, TypeVar, TypeVarTuple

T = TypeVar("T")
Ts = TypeVarTuple("Ts")

class Signal(Generic[*Ts, T]):
    def connect(
        self,
        callback: Callable[[*Ts], Any] | Callable[[*Ts, T], Any],
    ) -> Callable:
        return NotImplemented

# this signal emits up to 2 arguments, an int and string
signal: Signal[int, str] = Signal()

@signal.connect
def callback1(a: int, y: str): ...  # fine

@signal.connect
def callback2(a: int): ...  # ALSO fine

talley · March 30, 2024, 1:09pm

Thank you @Daverball, that is indeed very helpful! Indeed I don’t care about named parameters. I had a feeling TypeVarTuple could be useful but didn’t hit on this. I really do appreciate you taking a look.

If I may trouble you a bit further me now expose my real and fully annoying goals , which is to support this sort of syntax for up to N parameters with pre-defined overloads (somewhat like what @tiangolo is doing here with expressions in sqlmodel).

I’ve been playing with this for a while now with various strategies and still unsuccessful, so I’ll just paste the end-goal:

# I don't care whether Signal is a fake heavily overloaded factory function
sig2param = Signal(int, str) 
# or an actual generic class... whatever works
sig2param: Signal[int, str] = Signal() 

# Callbacks connected to this signal can take UP TO the specified N params

@sig2param.connect
def cb0() -> None: ...  # OK
@sig2param.connect
def cb0(a0: int) -> None: ...  # OK
@sig2param.connect
def cb0(a0: int, a1: str) -> None: ...  # OK

# but no more, and they must be the correct types:

@sig2param.connect
def cb0(a0: int, a1: str, a2: float) -> None: ...  # ERR: too many params
@sig2param.connect
def cb0(a0: str) -> None: ...  # ERR: wrong type for a0

Your solution with TypeVarTuple gets me most of the way there…

class Signal(Generic[*Ts, T]):
    def connect(self, cb: Callable[[*Ts], Any] | Callable[[*Ts, T], Any]): ...

but if I’m not mistaken, by adding the additional TypeVar T to the Generic, it’s hard to go backwards to support things that take fewer arguments?

# This will emit no arguments
sig0param = Signal()

If I omit the parameterization it naturally doesn’t like it:

sig0param: Signal = Signal()

@sig0param.connect
def callback4() -> None: ...

error: Argument 1 to "connect" of "Signal" has incompatible type "Callable[[], None]"; expected
"Callable[[VarArg(Any)], Any] | Callable[[VarArg(*tuple[*tuple[Any, ...], Any])], Any]"  [arg-type]
    @sig0param.connect
     ^
Found 1 error in 1 file (checked 1 source file)

so it seems i’m forced to add something like Any, which is wrong and leads to false negatives.

signal2: Signal[Any] = Signal()

@signal2.connect
def callback5() -> None: ...  # false negative. ERR: too many params

I know this is in the weeds on my specific use case, so thanks a lot for your time

talley · March 30, 2024, 1:12pm

i feel like pre-defined overloads on my side are the secret here, supporting up to N parameters. But can’t figure out how to pass the parameters on to the signature of the Callable passed to connect

Daverball · March 30, 2024, 3:30pm

Overloads could indeed get you something that works up to the Nth parameter, TypeVarTuple can only really give you a fixed number of optional parameters, type var defaults^[1] could potentially help you put this all into a single generic, but it will probably end up being not much more pretty or ergonomic than a bunch of overloads. Something fully generic that works for every possible case will be pretty much impossible with what we currently have available to us. At least I can’t think of a way to do it.

You could do something like this for up to N, just keep adding more overloads using the same strategy. I’m using PEP-695 syntax for brevity

from collections.abc import Callable
from typing import Any, overload

type Decorator[T] = Callable[[T], Any]

@overload
def Signal() -> Decorator[Callable[[], Any]]: ...
@overload
def Signal[T1](t1: type[T1]) -> Decorator[
    Callable[[], Any] | Callable[[T1], Any]
]: ...
@overload
def Signal[T1, T2](t1: type[T1], t2: type[T2]) -> Decorator[
    Callable[[], Any] | Callable[[T1], Any] | Callable[[T1, T2], Any]
]: ...

defaulting to typing.Never ↩︎

talley · March 30, 2024, 9:26pm

thanks, unfortunately, it’s not the Signal itself that is the decorator, it’s the method .connect on the Signal… still playing with various overload patterns, but haven’t hit on one that lets me pass on the parameters to the method. Thanks for your time though, much appreciated!

Daverball · March 30, 2024, 9:46pm

You can use the same pattern to bind a union of callables to Signal if you need it to be a class with a method:

from collections.abc import Callable
from typing import Any, overload

class Signal[FuncT: Callable[..., Any]]:
    @overload
    def __init__(self: Signal[Callable[[], Any]]) -> None: ...
    @overload
    def __init__[T1](
        self: Signal[Callable[[], Any] | Callable[[T1], Any]],
        t1: type[T1]
    ) -> None: ...
    @overload
    def __init__[T1, T2](
        self: Signal[
            Callable[[], Any]
            | Callable[[T1], Any]
            | Callable[[T1, T2], Any]
        ],
        t1: type[T1],
        t2: type[T2]
    ) -> None: ...

    def connect(self, func: FuncT): ...

Alternatively you can keep parametrization of Signal simple using a single TypeVarTuple and have the same overloads as above on the connect method and use the appropriate self parameter there, i.e. Signal[()] for the first overload Signal[T1] for the second, etc.

from collections.abc import Callable
from typing import Any, overload

type Decorator[T] = Callable[[T], Any]

class Signal[*Ts]:
    @overload
    def connect(self: Signal[()], func: Callable[[], Any]): ...
    @overload
    def connect[T1](
        self: Signal[T1],
        func: Callable[[], Any] | Callable[[T1], Any]
    ): ...

talley · March 30, 2024, 10:31pm

class Signal[FuncT: Callable[..., Any]]:
    @overload
    def __init__[T1](
        self: Signal[Callable[[], Any] | Callable[[T1], Any]],
        t1: type[T1]
    ) -> None: ...

just, wow. I never would have guessed that the overloaded __init__ function could close on the Callable union and parametrize the class itself. That’s some brilliant stuff right there. I could kiss you. I’ve been playing with this on and off for a number of months now. Playground here doing exactly what I want.

Alternatively you can keep parametrization of Signal simple using a single TypeVarTuple and have the same overloads as above on the connect method and use the appropriate self parameter there, i.e. Signal[()] for the first overload Signal[T1] for the second, etc.

This one I struggled to get working in the playground. If you’re inclined and can make the following work, with that alternate syntax, i would be very curious/

sig2: Signal[int, str] = Signal(int, str) 

@sig2.connect
def cb1() -> None: ...  # OK
@sig2.connect
def cb2(a0: int) -> None: ...  # OK
@sig2.connect
def cb3(a0: int, a1: str) -> None: ...  # OK

def cb4(x: int) -> None: ...  # ERR: too many params
@sig2.connect
def cb5(a0: int, a1: str, a2: float) -> None: ...  # ERR: too many params
@sig2.connect
def cb6(a0: str) -> None: ...  # ERR: wrong type for a0

in any case: huge thanks for your time.

Daverball · March 30, 2024, 10:38pm

It works just fine for me with your examples: Pyright Playground

talley · March 30, 2024, 10:51pm

huh. So it does. apologies; not sure what I was doing wrong before.

Well, thanks again!

gregbrowndev · August 29, 2024, 11:01am

Just wanted to add a fairly common use case that Concatenate can’t deal with (AFAIK).

If Concatenate only supports positional types at the beginning of the signature, how do you handle args that require a default value in the wrapped function?

As a concrete example, I wrote a higher-order function to handle pagination to Google’s Text Search API, goes something like this (using Python 3.12’s PEP 695 syntax):

async def text_search(
    text: str,
    location: Location | None,
    radius: float | None,
    next_page_token: str | None = None,
) -> TextSearchResponse:
    // impl. omitted
    pass


class HasNextPageToken(typing.Protocol):
    nextPageToken: str | None


def paginated[**P, R: HasNextPageToken](
    func: ???,  # Will expand on the options below
    paginate: bool = True,
) -> typing.Callable[P, typing.AsyncIterator[R]]:
    """
    Wrap an async function to paginate through the results
    """

    async def paginated(
        *args: P.args, **kwargs: P.kwargs
    ) -> typing.AsyncIterator[R]:
        next_page_token = None
        while True:
            response = await func(
                *args, **kwargs, next_page_token=next_page_token
            )
            yield response
            next_page_token = response.nextPageToken
            if not next_page_token or not paginate:
                break

    return paginated


paginated_text_search = paginated(text_search)
async for result in paginated_text_search(text="foo", ...):
    print(result)

If you type func as typing.Callable[typing.Concatenate[str| None, P], R], then you’d need to write text_search with next_page_token as a positional arg, which isn’t ideal as the default value should be None and args with default values need to go after the non-default args.

I also tried this approach. Mypy type checks paginated correct, but cannot seem to infer the type of the paginated(text_search) at all.

class PaginatableFunc[**P, R: HasNextPageToken](typing.Protocol):
    def __call__(
        *args: P.args, next_page_token: str | None = None , **kwargs: P.kwargs
    ) -> typing.Awaitable[R]:
        ...

where func: PaginatableFunc[P, R] in the paginated function.

alwaysmpe · November 19, 2024, 6:03pm

It’s possible to use 2 TypeVarTuples to achieve this for any arbitrary number of arguments using 3 overloads:

class Signal[*Ts]:
    @overload
    def connect(self, func: Callable[[], Any]) -> Any: ...
    @overload
    def connect[*Tc, *Tr](
        self: Signal[*Tc, *Tr],
        func: Callable[[*Tc], Any]
    ) -> Any: ...
    @overload
    def connect(
        self,
        func: Callable[[*Ts], Any]
    ) -> Any: ...

and this type checks in pyright, mypy rejects this however.

The PEP proposal said

This PEP was accepted for Python 3.11, with the caveat that details around multiple unpackings in a type expression aren’t specified precisely. This gives individual type checkers some leeway, but can be tightened in future PEPs.

Removing parameters from the *end* of a ParamSpec with Concatenate

Removing parameters from the end of a ParamSpec with Concatenate