Utilities#

yapw.util.json_dumps(message)[source]#

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)[source]#

If the content type is “application/json”, deserializes the JSON formatted bytes to a Python object. Otherwise, returns 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)[source]#

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)[source]#

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)[source]#

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

Parameters:
Returns:

arguments for logger.debug

Return type:

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