Method _encode_from_dict

gunicorn/dirty/protocol.py:576–643

Encode a message dict to binary format. Supports the old dict-based API for backwards compatibility. Args: message: Message dict with 'type', 'id', and payload fields Returns: bytes: Complete encoded message

(message: dict)

    @staticmethod
    def _encode_from_dict(message: dict) -> bytes:  # pylint: disable=too-many-return-statements
        """
        Encode a message dict to binary format.

        Supports the old dict-based API for backwards compatibility.

        Args:
            message: Message dict with 'type', 'id', and payload fields

        Returns:
            bytes: Complete encoded message
        """
        msg_type_str = message.get("type")
        request_id = message.get("id", 0)

        # Handle string or int request IDs
        if isinstance(request_id, str):
            # For backwards compat with UUID strings, hash to int
            request_id = hash(request_id) & 0xFFFFFFFFFFFFFFFF

        msg_type = MSG_TYPE_FROM_STR.get(msg_type_str)
        if msg_type is None:
            raise DirtyProtocolError(f"Unknown message type: {msg_type_str}")

        if msg_type == MSG_TYPE_REQUEST:
            return BinaryProtocol.encode_request(
                request_id,
                message.get("app_path", ""),
                message.get("action", ""),
                message.get("args"),
                message.get("kwargs")
            )
        elif msg_type == MSG_TYPE_RESPONSE:
            return BinaryProtocol.encode_response(
                request_id,
                message.get("result")
            )
        elif msg_type == MSG_TYPE_ERROR:
            return BinaryProtocol.encode_error(
                request_id,
                message.get("error", {})
            )
        elif msg_type == MSG_TYPE_CHUNK:
            return BinaryProtocol.encode_chunk(
                request_id,
                message.get("data")
            )
        elif msg_type == MSG_TYPE_END:
            return BinaryProtocol.encode_end(request_id)
        elif msg_type == MSG_TYPE_STASH:
            return BinaryProtocol.encode_stash(
                request_id,
                message.get("op"),
                message.get("table", ""),
                message.get("key"),
                message.get("value"),
                message.get("pattern")
            )