kiwipy package

Robust, high-volume, message based communication made easy.

class kiwipy.BroadcastFilter(subscriber, subject=None, sender=None)

Bases: object

A filter that can be used to limit the subjects and/or senders that will be received

classmethod _ensure_filter(filter_value)

classmethod _make_regex(filter_str)

Parameters

filter_str (str) – The filter string

Returns

The regular expression object

add_sender_filter(sender_filter)

add_subject_filter(subject_filter)

is_filtered(sender, subject) → bool

class kiwipy.Communicator

Bases: object

The interface for a communicator used to both send and receive various types of message.

abstract add_broadcast_subscriber(subscriber: Callable[[Communicator, Any, Any, Any, Any], Any], identifier=None) → Any

Add a broadcast subscriber that will receive all broadcast messages

Parameters

• subscriber – the subscriber function to be called

• identifier – an optional identifier for the subscriber

Returns

an identifier for the subscriber and can be subsequently used to remove it

abstract add_rpc_subscriber(subscriber: Callable[[Communicator, Any], Any], identifier=None) → Any

Add an RPC subscriber to the communicator with an optional identifier. If an identifier is not provided the communicator will generate a unique one. In all cases the identifier will be returned.

abstract add_task_subscriber(subscriber: Callable[[Communicator, Any], Any], identifier=None) → Any

Add a task subscriber to the communicator’s default queue. Returns the identifier.

Parameters

• subscriber – The task callback function

• identifier – the subscriber identifier

abstract broadcast_send(body, sender=None, subject=None, correlation_id=None) → bool

Broadcast a message to all subscribers

abstract close()

Close a communicator, free up all resources and do not allow any further operations

abstract is_closed() → bool

Return True if the communicator was closed

abstract remove_broadcast_subscriber(identifier)

Remove a broadcast subscriber

Parameters

identifier – the identifier of the subscriber to remove

abstract remove_rpc_subscriber(identifier)

Remove an RPC subscriber given the identifier. Raises a ValueError if there is no such subscriber.

Parameters

identifier – The RPC subscriber identifier

abstract remove_task_subscriber(identifier)

Remove a task subscriber from the communicator’s default queue.

Parameters

identifier – the subscriber to remove

Raises

ValueError if identifier does not correspond to a known subscriber

abstract rpc_send(recipient_id, msg)

Initiate a remote procedure call on a recipient. This method returns a future representing the outcome of the call.

Parameters

• recipient_id – The recipient identifier

• msg – The body of the message

Returns

A future corresponding to the outcome of the call

Return type

kiwipy.Future

abstract task_send(task, no_reply=False) → Future

Send a task messages, this will be queued and picked up by a worker at some point in the future. The method returns a future representing the outcome of the task.

Parameters

• task – The task message

• no_reply (bool) – Do not send a reply containing the result of the task

Returns

A future corresponding to the outcome of the task

exception kiwipy.CommunicatorClosed

Bases: Exception

Raised when an operation is attempted on a closed communicator

class kiwipy.CommunicatorHelper

Bases: Communicator

_ensure_open()

add_broadcast_subscriber(subscriber: Callable[[Communicator, Any, Any, Any, Any], Any], identifier=None) → Any

Add a broadcast subscriber that will receive all broadcast messages

Parameters

• subscriber – the subscriber function to be called

• identifier – an optional identifier for the subscriber

Returns

an identifier for the subscriber and can be subsequently used to remove it

add_rpc_subscriber(subscriber, identifier=None) → Any

Add an RPC subscriber to the communicator with an optional identifier. If an identifier is not provided the communicator will generate a unique one. In all cases the identifier will be returned.

add_task_subscriber(subscriber, identifier=None)

Register a task subscriber

Parameters

• subscriber – The task callback function

• identifier – the subscriber identifier

close()

Close a communicator, free up all resources and do not allow any further operations

fire_broadcast(body, sender=None, subject=None, correlation_id=None)

fire_rpc(recipient_id, msg)

fire_task(msg, no_reply=False)

is_closed() → bool

Return True if the communicator was closed

remove_broadcast_subscriber(identifier)

Remove a broadcast subscriber

Parameters

identifier – the identifier of the subscriber to remove

remove_rpc_subscriber(identifier)

Remove an RPC subscriber given the identifier. Raises a ValueError if there is no such subscriber.

Parameters

identifier – The RPC subscriber identifier

remove_task_subscriber(identifier)

Remove a task subscriber

Parameters

identifier – the subscriber to remove

Raises

ValueError if identifier does not correspond to a known subscriber

exception kiwipy.DeliveryFailed

Bases: Exception

Failed to deliver a message

exception kiwipy.DuplicateSubscriberIdentifier

Bases: Exception

Failed to add a subscriber because the identifier supplied is already in use

class kiwipy.EventHelper(listener_type)

Bases: object

add_listener(listener)

fire_event(event_function, *args, **kwargs)

property listeners

remove_all_listeners()

remove_listener(listener)

exception kiwipy.InvalidStateError

Bases: Exception

Raise when an object is in an invalid state to perform the desired action

class kiwipy.LocalCommunicator

Bases: CommunicatorHelper

broadcast_send(body, sender=None, subject=None, correlation_id=None)

Broadcast a message to all subscribers

rpc_send(recipient_id, msg)

Initiate a remote procedure call on a recipient. This method returns a future representing the outcome of the call.

Parameters

• recipient_id – The recipient identifier

• msg – The body of the message

Returns

A future corresponding to the outcome of the call

Return type

kiwipy.Future

task_send(task, no_reply=False)

Send a task messages, this will be queued and picked up by a worker at some point in the future. The method returns a future representing the outcome of the task.

Parameters

• task – The task message

• no_reply (bool) – Do not send a reply containing the result of the task

Returns

A future corresponding to the outcome of the task

exception kiwipy.QueueEmpty

Bases: Exception

Could not get the next message from the queue because it is empty

exception kiwipy.RemoteException

Bases: Exception

An exception occurred at the remote end of the call

exception kiwipy.TaskRejected

Bases: Exception

A task was rejected at the remote end

exception kiwipy.UnroutableError

Bases: DeliveryFailed

The messages was unroutable

kiwipy.capture_exceptions(future, ignore=())

Capture any exceptions in the context and set them as the result of the given future

Parameters

• future (kiwipy.Future) – The future to the exception on

• ignore – An optional list of exception types to ignore, these will be raised and not set on the future

kiwipy.chain(source, target)

Chain two futures together so that when one completes, so does the other.

The result (success or failure) of a will be copied to b, unless b has already been completed or cancelled by the time a finishes.

kiwipy.connect(uri: str = 'amqp://guest:guest@127.0.0.1/', **kwargs)

Create a connection using a URI

kiwipy.copy_future(source, target)

Copy the status of future a to b unless b is already done in which case return

Parameters

• source (kiwipy.Future) – The source future

• target (kiwipy.Future) – The target future

Subpackages

• kiwipy.rmq package
  • RmqCommunicator
    • RmqCommunicator._connection
    • RmqCommunicator._default_task_queue
    • RmqCommunicator._ensure_connected()
    • RmqCommunicator._message_publisher
    • RmqCommunicator._message_subscriber
    • RmqCommunicator.add_broadcast_subscriber()
    • RmqCommunicator.add_close_callback()
    • RmqCommunicator.add_rpc_subscriber()
    • RmqCommunicator.add_task_subscriber()
    • RmqCommunicator.broadcast_send()
    • RmqCommunicator.connect()
    • RmqCommunicator.connected()
    • RmqCommunicator.disconnect()
    • RmqCommunicator.get_default_task_queue()
    • RmqCommunicator.get_message_publisher()
    • RmqCommunicator.get_message_subscriber()
    • RmqCommunicator.loop
    • RmqCommunicator.remove_broadcast_subscriber()
    • RmqCommunicator.remove_rpc_subscriber()
    • RmqCommunicator.remove_task_subscriber()
    • RmqCommunicator.rpc_send()
    • RmqCommunicator.server_properties
    • RmqCommunicator.task_queue()
    • RmqCommunicator.task_send()
  • RmqIncomingTask
    • RmqIncomingTask._finalise()
    • RmqIncomingTask._on_task_done()
    • RmqIncomingTask._outcome_destroyed()
    • RmqIncomingTask._task_done()
    • RmqIncomingTask.body
    • RmqIncomingTask.no_reply
    • RmqIncomingTask.process()
    • RmqIncomingTask.processing()
    • RmqIncomingTask.requeue()
    • RmqIncomingTask.state
  • RmqTaskPublisher
    • RmqTaskPublisher.task_send()
  • RmqTaskQueue
    • RmqTaskQueue.add_task_subscriber()
    • RmqTaskQueue.connect()
    • RmqTaskQueue.disconnect()
    • RmqTaskQueue.next_task()
    • RmqTaskQueue.remove_task_subscriber()
    • RmqTaskQueue.task_send()
  • RmqTaskSubscriber
    • RmqTaskSubscriber.TASK_QUEUE_ARGUMENTS
    • RmqTaskSubscriber._build_response_message()
    • RmqTaskSubscriber._create_task_queue()
    • RmqTaskSubscriber._on_task()
    • RmqTaskSubscriber._send_response()
    • RmqTaskSubscriber.add_task_subscriber()
    • RmqTaskSubscriber.connect()
    • RmqTaskSubscriber.next_task()
    • RmqTaskSubscriber.remove_task_subscriber()
  • RmqThreadCommunicator
    • RmqThreadCommunicator.TASK_TIMEOUT
    • RmqThreadCommunicator._ensure_open()
    • RmqThreadCommunicator._wrap_future()
    • RmqThreadCommunicator._wrap_subscriber()
    • RmqThreadCommunicator.add_broadcast_subscriber()
    • RmqThreadCommunicator.add_close_callback()
    • RmqThreadCommunicator.add_rpc_subscriber()
    • RmqThreadCommunicator.add_task_subscriber()
    • RmqThreadCommunicator.broadcast_send()
    • RmqThreadCommunicator.close()
    • RmqThreadCommunicator.connect()
    • RmqThreadCommunicator.is_closed()
    • RmqThreadCommunicator.loop()
    • RmqThreadCommunicator.remove_broadcast_subscriber()
    • RmqThreadCommunicator.remove_rpc_subscriber()
    • RmqThreadCommunicator.remove_task_subscriber()
    • RmqThreadCommunicator.rpc_send()
    • RmqThreadCommunicator.server_properties
    • RmqThreadCommunicator.start()
    • RmqThreadCommunicator.stop()
    • RmqThreadCommunicator.task_queue()
    • RmqThreadCommunicator.task_send()
  • RmqThreadTaskQueue
    • RmqThreadTaskQueue.add_task_subscriber()
    • RmqThreadTaskQueue.next_task()
    • RmqThreadTaskQueue.remove_task_subscriber()
    • RmqThreadTaskQueue.task_send()
  • async_connect()
  • connect()
  • Submodules
    • kiwipy.rmq.communicator module
      • RmqCommunicator
      • async_connect()
    • kiwipy.rmq.configure module
    • kiwipy.rmq.defaults module
    • kiwipy.rmq.messages module
      • BaseConnectionWithExchange
      • BasePublisherWithReplyQueue
      • BroadcastMessage
    • kiwipy.rmq.tasks module
      • RmqIncomingTask
      • RmqTaskPublisher
      • RmqTaskQueue
      • RmqTaskSubscriber
    • kiwipy.rmq.threadcomms module
      • RmqThreadCommunicator
      • RmqThreadTaskQueue
      • connect()
    • kiwipy.rmq.utils module

Submodules

• kiwipy.communicate module
  • connect()
• kiwipy.communications module
  • Communicator
    • Communicator.add_broadcast_subscriber()
    • Communicator.add_rpc_subscriber()
    • Communicator.add_task_subscriber()
    • Communicator.broadcast_send()
    • Communicator.close()
    • Communicator.is_closed()
    • Communicator.remove_broadcast_subscriber()
    • Communicator.remove_rpc_subscriber()
    • Communicator.remove_task_subscriber()
    • Communicator.rpc_send()
    • Communicator.task_send()
  • CommunicatorHelper
    • CommunicatorHelper._ensure_open()
    • CommunicatorHelper.add_broadcast_subscriber()
    • CommunicatorHelper.add_rpc_subscriber()
    • CommunicatorHelper.add_task_subscriber()
    • CommunicatorHelper.close()
    • CommunicatorHelper.fire_broadcast()
    • CommunicatorHelper.fire_rpc()
    • CommunicatorHelper.fire_task()
    • CommunicatorHelper.is_closed()
    • CommunicatorHelper.remove_broadcast_subscriber()
    • CommunicatorHelper.remove_rpc_subscriber()
    • CommunicatorHelper.remove_task_subscriber()
• kiwipy.exceptions module
  • CommunicatorClosed
  • DeliveryFailed
  • DuplicateSubscriberIdentifier
  • InvalidStateError
  • QueueEmpty
  • RemoteException
  • TaskRejected
  • UnroutableError
• kiwipy.filters module
  • BroadcastFilter
    • BroadcastFilter._ensure_filter()
    • BroadcastFilter._make_regex()
    • BroadcastFilter.add_sender_filter()
    • BroadcastFilter.add_subject_filter()
    • BroadcastFilter.is_filtered()
• kiwipy.futures module
  • capture_exceptions()
  • chain()
  • copy_future()
• kiwipy.local module
  • LocalCommunicator
    • LocalCommunicator.broadcast_send()
    • LocalCommunicator.rpc_send()
    • LocalCommunicator.task_send()
• kiwipy.utils module
  • EventHelper
    • EventHelper.add_listener()
    • EventHelper.fire_event()
    • EventHelper.listeners
    • EventHelper.remove_all_listeners()
    • EventHelper.remove_listener()