Shared via mypy Playground

main.py

from typing import TypeVar, Self, Type, Protocol, assert_never, reveal_type


class MyRequest: ...


class MyResponse:
    @classmethod
    def from_bytes(cls: Type[Self], frame: bytes) -> Self:
        # deserialize ...
        return cls()


class MyError(MyResponse): ...


class StatusRequest(MyRequest): ...
class StatusResponse(MyResponse): ...
class StatusError(MyError): ...
class Status(StatusRequest):
    _Response = StatusResponse
    _Error = StatusError


class JobRequest(MyRequest): ...
class JobResponse(MyResponse): ...
class JobError(MyError): ...
class Job(JobRequest):
    _Response = JobResponse
    _Error = JobError


TResponse = TypeVar("TResponse", bound=MyResponse)
TError = TypeVar("TError", bound=MyError)


class Request(Protocol[TResponse, TError]):
    _Response: Type[TResponse]
    _Error: Type[TError]


async def request(request: Request[TResponse, TError]) -> TResponse | TError:
    # ... do some side effects over a transport
    is_error = False  # would be determined by what the transport does
    frame: bytes = b""  # would actually be read from the transport

    if is_error:
        return request._Error.from_bytes(frame)
    else:
        return request._Response.from_bytes(frame)


async def main() -> None:
    match status := await request(Status()):
        case StatusResponse():
            reveal_type(status)
            print(f"{status.__class__.__name__=}")
        case StatusError():
            reveal_type(status)
            print(f"{status.__class__.__name__=}")
        case _:
            assert_never(status)

    match job := await request(Job()):
        case JobResponse():
            reveal_type(job)
            print(f"{job.__class__.__name__=}")
        case JobError():
            reveal_type(job)
            print(f"{job.__class__.__name__=}")
        case _:
            assert_never(job)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())