Utilities

yapw.util.json_dumps(message)

Serialize a Python object to JSON formatted bytes.

Uses orjson if available.

Parameters:

message (Any) – a Python object

Returns:

JSON formatted bytes

Return type:

bytes

yapw.util.default_decode(body, content_type)

If the content type is “application/json”, deserialize the JSON formatted bytes to a Python object.

Otherwise, return the bytes (which the consumer callback can deserialize independently).

Uses orjson if available.

Parameters:

• body (bytes) – the encoded message

• content_type (str | None) – the message’s content type

Returns:

a Python object

Return type:

Any

yapw.util.default_encode(message, content_type)

Encode the decoded message to bytes.

• If the content type is “application/json”, serialize the message to JSON formatted bytes.

• If the message is a string, encode it to bytes.

• If the message is bytes, return it.

• Otherwise, serialize the message to its picked representation.

Parameters:

• message (Any) – a decoded message

• content_type (str) – the message’s content type

Returns:

an encoded message

Return type:

bytes

yapw.util.basic_publish_kwargs(state, message, routing_key)

Prepare keyword arguments for basic_publish.

Parameters:

• state (Base[Any] | State[Any]) – an object with the attributes format_routing_key, exchange, encode, content_type and delivery_mode

• message (Any) – a decoded message

• routing_key (str) – the routing key

Returns:

keyword arguments for basic_publish

Return type:

PublishKeywords

yapw.util.basic_publish_debug_args(channel, message, keywords)

Prepare arguments for logger.debug related to publishing a message.

Parameters:

• channel (Channel | BlockingChannel) – the channel from which to call basic_publish

• message (Any) – a decoded message

• keywords (PublishKeywords) – keyword arguments for basic_publish

Returns:

arguments for logger.debug

Return type:

tuple[str, Any, int, str, str]