xadd & xread & xreadgroup use invariant dict typing

[imnotjames]

Along with all of the other choices made by the previous maintainers around typing, it seems the mappings passed to xadd, read, and xreadgroup are invariant and do not allow passing in values like dict[str, str]. This makes sense when you're concerned about someone setting an incompatible value into the dict or mapping. A bit less for cases like this where we just want to read the values. (I think? I'm not an expert on Python typing!)

This can be seen with xreadgroup, when passing in the streams and stream IDs:

valkey-py/valkey/commands/core.py

Lines 3972 to 3980 in c490780

	def xreadgroup(
	self,
	groupname: str,
	consumername: str,
	streams: Dict[KeyT, StreamIdT],
	count: Union[int, None] = None,
	block: Union[int, None] = None,
	noack: bool = False,
	) -> ResponseT:

If I were to try to interact with this like such:

stream = { "incoming-text": "$" }
payload = cast(Any, client.xread(stream, None, 0))

'course, things like pyright don't like this and emit an error about it. Invariant types and all that.

Pyright: Argument of type "dict[str, str]" cannot be assigned to parameter "streams" of type "Dict[KeyT, Unknown]" in function "xread"
   "dict[str, str]" is not assignable to "Dict[KeyT, Unknown]"
     Type parameter "_KT@dict" is invariant, but "str" is not the same as "KeyT" (reportArgumentType)

There's a helpful comment about this in the typing.py file.

valkey-py/valkey/typing.py

Lines 42 to 46 in c490780

	# Mapping is not covariant in the key type, which prevents
	# Mapping[_StringLikeT, X] from accepting arguments of type Dict[str, X]. Using
	# a TypeVar instead of a Union allows mappings with any of the permitted types
	# to be passed. Care is needed if there is more than one such mapping in a
	# type signature because they will all be required to be the same key type.

Using this, we can apply it to StreamIdTto create a version that's covariant.

AnyStreamIdT = TypeVar("AnyStreamIdT", int, bytes, str, memoryview)

 def xreadgroup( 
     self, 
     groupname: str, 
     consumername: str, 
-     streams: Dict[KeyT, StreamIdT], 
+     streams: Mapping[AnyKeyT, AnyStreamIDT],
     count: Union[int, None] = None, 
     block: Union[int, None] = None, 
     noack: bool = False, 
 ) -> ResponseT: 

It seems to work well! I tested it with the above example with xreadgroup, at least. I can't seem to find any issues to this approach -- and it seems to be used elsewhere even if not here.

[imnotjames]

There's an alternative but it's definitely messier.

KT_co = TypeVar("KT_co", covariant=True)
VT_co = TypeVar("VT_co", covariant=True)

class CovariantMapping(Protocol[KT_co, VT_co]):
    def __eq__(self, other: object) -> bool: ...
    def __ne__(self, other: object) -> bool: ...
    def keys(self) -> KeysView[KT_co]: ...
    def items(self) -> ItemsView[KT_co, VT_co]: ...
    def values(self) -> ValuesView[VT_co]: ...

[imnotjames]

I have a draft PR up that I'm currently testing locally. Unfortunately, related to #84 (some really awesome effort there!) it's somewhat difficult to check types directly.

[mkmkme]

Thanks for your report and the draft PR! Yeah I was about to mention #84.
@Rafiot could you also take a look into the attached PR? Does it clash with yours?

[Rafiot]

First of all, sorry I've not been able to allocate more time to the #84.

#130 seems fine, I don't see any obvious issue with it (note that I'm not an expert in Python typing either). Once merged, I can rebase #84 and keep going (given time).

@imnotjames: if you're interesting in contributing to #84, I'd love that. There is a lot more work to do before it's mergeable.

[Rafiot]

Just FYI, I re-did the PR (#134), this time with signoff so it's somewhat closer to mergeable.

[amirreza8002]

hi @Rafiot
can you write some info about the state of this pr and what you need help with?
it's a bit hard to navigate a pr of this sort

[Rafiot]

@amirreza8002 It probably makes more sense to have this discussion in the PR, here it is: #134 (comment)