The pika-stubs from hahow

Incomplete type for BlockingConnection adapter init parameters

The type of the BlockingConnection.__init__ parameters param is Optional[connection.Parameters]:

pika-stubs/pika-stubs/adapters/blocking_connection.pyi

Lines 9 to 14 in 361dc1a

    
           class BlockingConnection: 
        
               def __init__( 
        
                   self, 
        
                   parameters: Optional[connection.Parameters] = ..., 
        
                   _impl_class: Optional[Any] = ...,

But according to pika docs it should be "Connection parameters instance or non-empty sequence of them" i.e. Union[pika.connection.Parameters, Sequence[pika.connection.Parameters]]

Incorrect return type of `consume()` method.

pika-stubs/pika-stubs/adapters/blocking_connection.pyi

Lines 143 to 150 in 361dc1a

    
           def consume( 
        
               self, 
        
               queue: str, 
        
               auto_ack: bool = ..., 
        
               exclusive: bool = ..., 
        
               arguments: Optional[Mapping[str, Any]] = ..., 
        
               inactivity_timeout: Optional[float] = ..., 
        
           ) -> None: ...

This returns a generator that yields tuple(pika.spec.Basic.Deliver, pika.spec.BasicProperties, bytes) so it should be:

typing.Iterator[tuple(pika.spec.Basic.Deliver, pika.spec.BasicProperties, bytes)]

basic_publish allows `str` type

basic_publish's body parameter, e.g.:

pika-stubs/pika-stubs/adapters/blocking_connection.pyi

Lines 171 to 178 in 361dc1a

    
           def basic_publish( 
        
               self, 
        
               exchange: str, 
        
               routing_key: str, 
        
               body: bytes, 
        
               properties: Optional[spec.BasicProperties] = ..., 
        
               mandatory: bool = ..., 
        
           ) -> None: ...

or

pika-stubs/pika-stubs/channel.pyi

Lines 110 to 117 in 361dc1a

    
           def basic_publish( 
        
               self, 
        
               exchange: str, 
        
               routing_key: str, 
        
               body: bytes, 
        
               properties: Optional[spec.BasicProperties] = ..., 
        
               mandatory: bool = ..., 
        
           ) -> None: ...

allows a str argument, per pika code:

   def basic_publish(self,
                     exchange,
                     routing_key,
                     body,
                     properties=None,
                     mandatory=False):
       """Publish to the channel with the given exchange, routing key and body.
       For more information on basic_publish and what the parameters do, see:

       http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish

       :param str exchange: The exchange to publish to
       :param str routing_key: The routing key to bind on
       :param bytes body: The message body
       :param pika.spec.BasicProperties properties: Basic.properties
       :param bool mandatory: The mandatory flag

       """
       self._raise_if_not_open()
       if isinstance(body, unicode_type):
           body = body.encode('utf-8')
       properties = properties or spec.BasicProperties()
       ...

Mismatch datatype of `consumer_tag` in `Basic.ConsumeOk`

This is how Pika generate consumer tag (line 336 in pika/channel.py):

def _generate_consumer_tag(self):
    """Generate a consumer tag

    NOTE: this protected method may be called by derived classes

    :returns: consumer tag
    :rtype: str

    """
    return 'ctag%i.%s' % (self.channel_number, uuid.uuid4().hex)

By default, the consumer tag is a str, which contradicts the data type of ConsumeOk.consume_tag in the stub file (line 1415 in pika-stubs/spec.pyi):

class ConsumeOk(amqp_object.Method):
    ...
    consumer_tag: Optional[int] = ...
    ...

So, is this an error on pika-stubs?

Disclaimer: I haven't read the AMQP specification. (Edit: I'm reading it, but not thoroughly)

Pip install fails

I have:

python 3.7.7
pip 20.2.4

Running

pip install pika-stubs

Fails with the following error:

$ pip install pika-stubs
ERROR: Could not find a version that satisfies the requirement pika-stubs (from versions: none)
ERROR: No matching distribution found for pika-stubs

Incorrect _DEFAULT typing for ConnectionParameters arguments

For the ConnectionParameters, the type are defined as follows:

class ConnectionParameters(Parameters):

    class _DEFAULT:
        ...

    def __init__(
        self,
        host: Union[str, _DEFAULT] = ...,
        port: Union[int, _DEFAULT] = ...,
        virtual_host: Union[str, _DEFAULT] = ...,
        credentials: Union[credentials_.VALID_TYPES, _DEFAULT] = ...,
        channel_max: Union[int, _DEFAULT] = ...,
        frame_max: Union[int, _DEFAULT] = ...,
        heartbeat: Union[int, _OnHeartbeatTimeoutCallback, None, _DEFAULT] = ...,
        ssl_options: Union[SSLOptions, _DEFAULT] = ...,
        connection_attempts: Union[int, _DEFAULT] = ...,
        retry_delay: Union[numbers.Real, _DEFAULT] = ...,
        socket_timeout: Union[numbers.Real, _DEFAULT] = ...,
        stack_timeout: Union[numbers.Real, _DEFAULT] = ...,
        locale: Union[str, _DEFAULT] = ...,
        blocked_connection_timeout: Union[numbers.Real, None, _DEFAULT] = ...,
        client_properties: Union[Mapping[str, Any], None, _DEFAULT] = ...,
        tcp_options: Union[Mapping[str, Any], None, _DEFAULT] = ...,
        **kwargs: Any,
    ): ...

The _DEFAULT type defined expects an instance, but the actual method only checks for the _DEFAULT type itself. I believe the _DEFAULT type should be changed to Type[_DEFAULT]:

class ConnectionParameters(Parameters):

    class _DEFAULT:
        ...

    def __init__(
        self,
        host: Union[str, Type[_DEFAULT]] = ...,
        port: Union[int, Type[_DEFAULT]] = ...,
        virtual_host: Union[str, Type[_DEFAULT]] = ...,
        credentials: Union[credentials_.VALID_TYPES, Type[_DEFAULT]] = ...,
        channel_max: Union[int, Type[_DEFAULT]] = ...,
        frame_max: Union[int, Type[_DEFAULT]] = ...,
        heartbeat: Union[int, _OnHeartbeatTimeoutCallback, None, Type[_DEFAULT]] = ...,
        ssl_options: Union[SSLOptions, Type[_DEFAULT]] = ...,
        connection_attempts: Union[int, Type[_DEFAULT]] = ...,
        retry_delay: Union[numbers.Real, Type[_DEFAULT]] = ...,
        socket_timeout: Union[numbers.Real, Type[_DEFAULT]] = ...,
        stack_timeout: Union[numbers.Real, Type[_DEFAULT]] = ...,
        locale: Union[str, Type[_DEFAULT]] = ...,
        blocked_connection_timeout: Union[numbers.Real, None, Type[_DEFAULT]] = ...,
        client_properties: Union[Mapping[str, Any], None, Type[_DEFAULT]] = ...,
        tcp_options: Union[Mapping[str, Any], None, Type[_DEFAULT]] = ...,
        **kwargs: Any,
    ): ...

Incorrect type for parameter 'arguments'.

pika-stubs/pika-stubs/adapters/blocking_connection.pyi

Line 232 in 361dc1a

arguments: Optional[spec.Queue.DeclareOk] = ...,

Should be:

arguments: Optional[Mapping[str, Any]] = ...,

Use of numbers.* encounters a bug in mypy

python/mypy#3186

Could Union[int, float] be used instead of numbers.Real?

hahow / pika-stubs Goto Github PK

pika-stubs's People

Stargazers

Watchers

Forkers

pika-stubs's Issues

Incomplete type for BlockingConnection adapter init parameters

Incorrect return type of `consume()` method.

basic_publish allows `str` type

Mismatch datatype of `consumer_tag` in `Basic.ConsumeOk`

Pip install fails

Incorrect _DEFAULT typing for ConnectionParameters arguments

Incorrect type for parameter 'arguments'.

Use of numbers.* encounters a bug in mypy

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent

	class BlockingConnection:

	def __init__(
	self,
	parameters: Optional[connection.Parameters] = ...,
	_impl_class: Optional[Any] = ...,

	def consume(
	self,
	queue: str,
	auto_ack: bool = ...,
	exclusive: bool = ...,
	arguments: Optional[Mapping[str, Any]] = ...,
	inactivity_timeout: Optional[float] = ...,
	) -> None: ...

	def basic_publish(
	self,
	exchange: str,
	routing_key: str,
	body: bytes,
	properties: Optional[spec.BasicProperties] = ...,
	mandatory: bool = ...,
	) -> None: ...