Typing / Mypy Support

[benjamin-kirkbride]

This project is amazing, thanks so much!

Are type annotations something that this project is open to having? I could try to take a stab at it, but I have never typed a C++ package before. I assume I would just make stubs?

[alex-petrenko]

Apologies for radio silence!
Yes, I believe that would be a fine contribution! I've never done this myself. Wonder if it would be enough to just type the faster_fifo.pyx file?

[leandrobbraga]

I managed to type annotate the faster_fifo.pyx, but didn't manage to export the types correctly to the library user.

The following code can be transformed in a .pyi file, or you can add these types to the .pyx itself:

import threading
from typing import Callable
from typing import Generic
from typing import Sequence
from typing import TypeVar

T = TypeVar("T")

class TLSBuffer(threading.local): ...

class Queue(Generic[T]):
    def __init__(
        self,
        max_size_bytes: int = ...,
        maxsize: int = ...,
        loads: Callable[[memoryview], T] | None = ...,
        dumps: Callable[[T], bytes] | None = ...,
    ) -> None: ...
    def close(self) -> None: ...
    def is_closed(self) -> bool: ...
    def put_many(self, xs: Sequence[T], block: bool = ..., timeout: float = ...) -> None: ...
    def put(self, x: T, block: bool = ..., timeout: float = ...) -> None: ...
    def put_many_nowait(self, xs: Sequence[T]) -> None: ...
    def put_nowait(self, x: T) -> None: ...
    def get_many(
        self, block: bool = ..., timeout: float = ..., max_messages_to_get: int = ...
    ) -> list[T]: ...
    def get_many_nowait(self, max_messages_to_get: int = ...) -> list[T]: ...
    def get(self, block: bool = ..., timeout: float = ...) -> T: ...
    def get_nowait(self) -> T: ...
    def qsize(self) -> int: ...
    def empty(self) -> bool: ...
    def full(self) -> bool: ...
    def data_size(self) -> int: ...
    def parse_messages(
        self, num_messages: int, total_bytes: int, msg_buffer: TLSBuffer
    ) -> list[T]: ...
    def reallocate_msg_buffer(self, new_size: int) -> None: ...
    def loads(self, msg_bytes: memoryview) -> T: ...
    def dumps(self, obj: T) -> bytes: ...
    def join_thread(self) -> None: ...
    def cancel_join_thread(self) -> None: ...

I tried both using py.typed and faster_fifo.pyi but failed. I think it's possible, I'm just not doing it correctly. For now I'm adding this stub to my own project.