Is pyright treating invariant as contravariant here?

[raceychan]

Hello guys, recently I have been struggling with type hinting decorators, and would really appreaciate some help.

1. The goal: I have a alot of registry logic in my app, where I use decorator to register functions, so that my code can call them later. the signatures of Registered functions should share similar interface, so that they can be called interchangably .
2. The problem:
   the register method is a decorator, receiving the function and return it. but according to eric traut, using a same generic type param in both function input and output would result it being invariance, meaning that decorated function should have the exact same type with the type param T.

For example:

In this case, since type param T is used in both the input and the ouput of the register function, test2 does not pass the type cheking.

[raceychan]

I would further complete my question here with a more realistic example.

type EventHandler[E] = ty.Callable[[E], ty.Awaitable[None]]


class EventBus[E: IEvent]:
    _subscribers: dict[type[E], list[EventHandler[E]]]

    def __init__(self, es: EventStore):
        self._es = es
        self._subscribers = defaultdict(list)

    async def dispatch(self, event: E):
        """
        Notify all subscribers of the event
        """
        for handler in self._subscribers[type(event)]:
            await handler(event)

    @classmethod
    def register[
        T: IEvent
    ](cls, handler: ty.Callable[[T], ty.Awaitable[None]]) -> ty.Callable[
        [T], ty.Awaitable[None]
    ]:
        cls._subscribers[T].append(handler) # pylance would complain this
        return handler

note that register[T: IEvent] won’t work, because it is not the same type variable as E

[raceychan]

I somehow managed to make a simpler, but working version

I’ll share with you guys just in case it might help

import typing as ty


class IQuery[R](ty.Protocol): ...


type QueryHandler[Q, R] = ty.Callable[[Q], ty.Coroutine[ty.Any, ty.Any, R]]


class MessageBus:
    query_handlers: ty.ClassVar[
        dict[type[IQuery[ty.Any]], QueryHandler[ty.Any, ty.Any]]
    ]

    def register_query_handler[
        Q, R
    ](self, handler: QueryHandler[Q, R]) -> QueryHandler[Q, R]:
        return handler

    async def send[R](self, query: IQuery[R]) -> R:
        return await self.query_handlers[type(query)](query)


bus = MessageBus()


# Example subclass of Query with a specific return type
class GetUserQuery(IQuery[str]):
    def __init__(self, user_id: int):
        self.user_id = user_id


@bus.register_query_handler
async def get_user_handler(query: GetUserQuery):
    # return f"Handling GetUserQuery for user_id: {query.user_id}"
    ...


async def main():
    r = await bus.send(GetUserQuery(user_id=123))
    print(r)

Notice that here MessageBus is not a generic class.

edits:

1. fix grammar

2. change descirption

[raceychan]

type Handler[T] = ty.Callable[[int, T], ty.Any]

class ClassA:
    handlers: dict[type[ty.Any], Handler[ty.Any]] = {}

    def extract_type[T](self, handler: Handler[T]) -> type[T]:
        return handler.__annotations__["exc"]

    def register[T: str](self, handler: Handler[T]) -> Handler[T]:
        t = self.extract_type(handler)
        self.handlers[t] = handler
        return handler


class substr(str): ...


class grandstr(str): ...


a: ClassA = ClassA()


@a.register
def test0(request: int, exc: str): ...
@a.register
def test1(request: int, exc: substr): ...
@a.register
def test2(request: int, exc: grandstr): ...

eventually i solved this problem by removing the type param from class, and bind it with the register function.