Skip to content

(Async)StreamCommands.xclaim type annotations seem to be incorrect. #197

@notypecheck

Description

@notypecheck

It's impossible to pass list[str] (or any other inner type) to xclaim(message_ids=...):

from valkey import Valkey

client = Valkey()


# message_ids: list[int | bytes | str | memoryview[int]] = ["1", "2", "3"]
message_ids = ["1", "2", "3"]
client.xclaim(
    "name",
    "groupname",
    "consumername",
    min_idle_time=0,
    message_ids=message_ids
)
main.py:13:17: error: Argument "message_ids" to "xclaim" of "StreamCommands" has incompatible type "list[str]"; expected "list[int | bytes | str | memoryview[int]] | tuple[int | bytes | str | memoryview[int]]"  [arg-type]
        message_ids=message_ids
                    ^~~~~~~~~~~
main.py:13:17: note: "List" is invariant -- see https://mypy.readthedocs.io/en/stable/common_issues.html#variance
main.py:13:17: note: Consider using "Sequence" instead, which is covariant
Found 1 error in 1 file (checked 1 source file)

It type checks correctly if we explicitly annotate it with list[int | bytes | str | memoryview[int]]

More importantly second part of the union seem to be incorrect:

message_ids: Union[List[StreamIdT], Tuple[StreamIdT]],

Tuple[T] would be a tuple with a single element, it was probably meant to be a Tuple[StreamIdT, ...] to accept tuples with any number of elements.

Potential fixes

  • Since internally message_ids are only used as Iterable object it should be safe to change it's type to Sequence[T] or Iterable[T].
  • Make the function accept a homogeneous list, i.e. list[str] | list[bytes] | list[int] | ...

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions