How do I assert that observe takes a queue that I can add As to, without necessarily caring what else the queue could hold?

I have an asyncio.Queue with items of either A or B type. I therefore typed my class like this:

class Observer:
   def __init__(self):
       self.queue: Queue[A|B] = asyncio.Queue()

There is another method that takes the queue and calls queue.put with an A item. I'd like to type hint like this, but it fails:

def observe(queue: Queue[A]):
    ... # does stuff
    item = A(...)
    queue.put(item)

Now, Queue[A] is incompatible with Queue[A|B], obviously. How can I type hint the queue parameter such that it takes any Queue that accepts A items, regardless of other types it also accepts?

For what is worth, I already tried Protocols like this. And it fails saying A is incompatible with A|B.

class CanPutA(Protocol):
    async def put(self, item: A):
        ...

Your CanPutA can work if you make it generic and contravariant, but see below.

T = TypeVar('T', contravariant=True)

class CanPut(Protocol[T]):
    async def put(self, item: T):
        ...


 def observe(queue: CanPut[A]):
     item = A(...)
     queue.put(A)

Contravariance allows Queue[A|B] as an argument, because Queue[A] is a subclass of Queue[A|B]. (Ordinarily, generics are invariant, which means your CanPutA only accepted containers of A exactly, not sub- or superclasses of A.)

(In fact, if you didn't know about contravariance and just tried the generic protocol, mypy will tell you here that you are using an invariant type variable where a contravariant one is expected.

It's a quirk of mypy that containers are co-, contra-, or invariant, but you make that declaration on the type variables used to define them instead of on the generic itself.)

There is a caveat: observe no longer is restricted to Queues, but any class with an appropriate put method. That's probably OK. What mypy (at least) lacks is support for intersection types, which would allow you to say that queue must be both a Queue of some kind and something that supports CanPut[A].

 # Hypothetical
 def observe(queue: Intersection[Queue, CanPut[A]]):
     ...

Intersection types are on the radar of the mypy developers, but a combination of the perceived importance and estimated amount of work to implement correctly has prevented it from being supported to date.