>source

웹 소켓 스트림에 메시지를 보내고 비슷한 데이터를 포함하는 메시지를 다시받는 두 가지 스키마가 있다고 가정 해 봅시다.

# First Schema
x_sent = {"Product": {"id": "123"}}
x_received = {"properties": {
    "id": {"type": "number"},
    "color": "green"}}
# Second Schema
y_sent = {"Item": {"Product": {"uid": "123"}}}
y_received = {"configs": {
    "id_number": "123",
    "type": "int"},
    "colour": "green"}

두 스트림을 구별하려면 메시지 내용을 필터링 할 수 있습니다.

if msg == "properties":
    use_schema_a()
if msg == "configs":
    use_schema_b()

그러나 다른 스키마의 양이 커지면 그렇게 건조하지 않습니다. 다음과 같이 할 수도 있습니다 :

msg_routing = {"properties": use_schema_a,
               "configs": use_schema_b}
if msg:
    msg_routing[msg]()

그러나 여전히 모든 스키마에 대해 함수를 만들고 있습니다! 나는 (개념적으로) 무언가를 놓친 것처럼 느낍니다. 메시지 전송 및 수신을 처리하고 구성 파일 형식의 스트림 별 필터링 데이터 만있는 일반 클래스를 만들고 싶습니다.

다음과 같이 보일 수 있습니다 :

{"schemaA": {"name": "service_ABC", "color": "properties.color", "send_id":"Product:id"},
 "schemaB": {"name": "service_DEF", "color": "configs.colour", "send_id":"Item:Product:uid"}}

위의 예와 마찬가지로 필요한 데이터는 동일합니다 ( green  이 예에서). ID  데이터가 비슷해 지도록 보내야합니다 ( 123  이 예에서).

따라서 보내고 받아야하는 데이터의 스키마를 알고 있다면 해당 스키마를 이해하는 것을 어떻게 동적으로 구축합니까?

시작점에 대한 명확한 예를 제공하려면 :

def on_message(received_msg):
    # The unparsed message we receive is something like
    #      {"properties": {
    #     "id": {"type": "number"},
    #     "color": "green"}}
    # Do our message filtering/parsing
    handle_message_contents(service_name, color)

  • 답변 # 1

    먼저 스키마를 사용하여 메시지를 작성해야합니다. 예를 들어 다음과 같이 스키마를 만드는 경우 :

    schemas = {
      "service_ABC": {
        "send": {
          "id": ["Product", "id"],
        },
        "receive": {
          "color": ["properties", "color"],
        },  
      },
      "service_DEF": {
        "send": {
          "id": ["Item", "Product", "uid"],
          "cond": ["Item", "Condition"],
        },
        "receive": {
          "color": ["configs", "colour"],
        },
      },
    }
    
    

    서비스 이름과 올바른 인수가 제공 될 때 전송 될 데이터 사전을 구축 할 수있는 방법을 사용할 수 있습니다 :

    def build_request(service, **kwargs):
      request = dict()
      for attribute, path in schemas[service]["send"].items():
        second_to_last_level = request
        last_level = request
        for level in path:
          second_to_last_level = last_level
          last_level = last_level.setdefault(level, dict())
        second_to_last_level[level] = kwargs[attribute]
      return request
    
    

    이 방법으로 다른 매개 변수를 추가하여 스키마로 직접 보낼 수 있습니다. 몇 가지 예를 참조하십시오 :

    build_request("service_ABC", id="123") == {
      "Product": {
        "id": "123"
      }
    }
    build_request("service_DEF", id="123", cond="New") == {
      "Item": {
        "Product": {
          "uid": "123"
        },
        "Condition": "New"
      }    
    }
    
    

    다음으로, 메시지가 어디에서 왔는지 식별해야합니다. 이를 수행하는 가장 좋은 방법은 업스트림 어딘가에 있고이를 "스키마 프로세서"로 전달하는 것입니다. 메시지와 함께 해당 정보를 얻을 방법이없는 경우 (의문의 여지가 있음) 제안 된 방법 중 하나를 사용할 수 있습니다.

    메시지가 어떤 서비스에서 왔는지, 따라서 어떤 스키마를 사용해야하는지 알게되면 요청을 작성하는 것과 비슷한 방식으로 메시지를 처리 ​​할 수 ​​있습니다.

    def process(service, msg):
      result = dict()
      for attribute, path in schemas[service]["receive"].items():
        value = msg
        for field in path:
          value = value[field]
        result[attribute] = value
      return result
    
    

    다시 예를보십시오 :

    x_received = {
      "properties": {
        "id": {
          "type": "number"
        },
        "color": "green"
      }
    }
    process("service_ABC", x_received) == {
      "color": "green"
    }
    
    

    정말로 service 를 지킬 수 없다면   process() 에 전달할 변수 그런 다음 msg_routing 를 사용하는 것이 가장 좋은 방법이라고 생각합니다. . 이것을 별도의 사전으로 사용하거나 schemas 에 추가 할 수도 있습니다 . 또는 항상 process() 를 체크인 할 수 있습니다  예상 한 것을 얻었는지 여부와 다음 스키마를 적용하지 않으려는 경우 :

    def process(msg):
      for service, schema in schemas.items():
        missing_something = False
        result = dict()
        for attribute, path in schema["receive"].items():
          value = msg
          for field in path:
            if not field in value:
              missing_something = True
              break
            value = value[field]
          if missing_something:
            break
          result[attribute] = value
        if not missing_something:
          return service, result
      raise RuntimeError("No schema applies")
    
    
    

  • 이전 python - str과 float가 필요할 때 bool과 int가 유형 오류로 간주되지 않는 이유는 무엇입니까?
  • 다음 python - numpy를 사용하여 상위 n 값을 제외하고 행렬 (2D 배열)의 모든 값을 제로화하는 방법은 무엇입니까?