消息 Schema 与角色边界

mkdir l6-messages
cd l6-messages

"""一个应用内部消息契约与请求配对器。"""
from copy import deepcopy

FIELDS = {"id", "task_id", "sender", "recipient", "type", "reply_to", "payload"}
ALLOWED = {"coordinator": {"request"}, "reader": {"result", "error"}}

def validate(message, authenticated_sender):
    if not isinstance(message, dict) or set(message) != FIELDS:
        raise ValueError("消息字段不完整")
    for key in ("id", "task_id", "sender", "recipient"):
        if not isinstance(message[key], str) or not message[key].strip():
            raise ValueError("编号和角色必须是非空字符串")
    if message["sender"] != authenticated_sender:
        raise PermissionError("信封发送者与认证身份不一致")
    if message["type"] not in ALLOWED.get(authenticated_sender, set()):
        raise PermissionError("该角色不能发送此消息类型")
    if not isinstance(message["payload"], dict):
        raise ValueError("payload必须是对象")
    if message["type"] == "request":
        if message["reply_to"] is not None or not isinstance(message["payload"].get("query"), str):
            raise ValueError("请求必须携带query且无reply_to")
    else:
        key = "value" if message["type"] == "result" else "code"
        if not isinstance(message["reply_to"], str) or key not in message["payload"]:
            raise ValueError("回复缺少配对编号或内容")

class Inbox:
    def __init__(self):
        self.pending, self.received, self.results = {}, {}, {}

    def add_request(self, message):
        validate(message, "coordinator")
        if message["id"] in self.pending:
            raise ValueError("请求编号重复")
        self.pending[message["id"]] = deepcopy(message)

    def receive(self, message, authenticated_sender):
        validate(message, authenticated_sender)
        previous = self.received.get(message["id"])
        if previous is not None:
            if previous != message:
                raise ValueError("同一消息编号对应不同内容")
            return "duplicate"
        request = self.pending.get(message["reply_to"])
        if request is None:
            raise ValueError("找不到配对请求")
        if (message["task_id"] != request["task_id"]
                or message["sender"] != request["recipient"]
                or message["recipient"] != request["sender"]):
            raise ValueError("回复任务或角色不匹配")
        if message["reply_to"] in self.results:
            raise ValueError("请求已收到最终回复")
        self.received[message["id"]] = deepcopy(message)
        self.results[message["reply_to"]] = deepcopy(message["payload"])
        return "accepted"

def request(number, query):
    return {"id": number, "task_id": "T-17", "sender": "coordinator",
            "recipient": "reader", "type": "request", "reply_to": None,
            "payload": {"query": query}}

def reply(number, request_id, value):
    return {"id": number, "task_id": "T-17", "sender": "reader",
            "recipient": "coordinator", "type": "result", "reply_to": request_id,
            "payload": {"value": value}}

def main():
    inbox = Inbox()
    inbox.add_request(request("Q-order", "订单金额"))
    inbox.add_request(request("Q-policy", "退款规则"))
    for message in [reply("M-policy", "Q-policy", "未发货可退"),
                    reply("M-order", "Q-order", 120)]:
        print(message["reply_to"] + "：" + inbox.receive(message, "reader"))
    print("按请求取回：金额=" + str(inbox.results["Q-order"]["value"]))
    print("重复消息：" + inbox.receive(reply("M-order", "Q-order", 120), "reader"))
    try:
        inbox.receive(reply("M-fake", "Q-order", 999), "coordinator")
    except PermissionError as error:
        print("伪造发送者：" + str(error))

if __name__ == "__main__":
    main()

ls
python -X utf8 messages_demo.py

Q-policy：accepted
Q-order：accepted
按请求取回：金额=120
重复消息：duplicate
伪造发送者：信封发送者与认证身份不一致

python -X utf8 -m unittest -v test_messages_demo.py