#!/usr/bin/env python ## # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Protocol implementation.""" from __future__ import absolute_import, division, print_function import hashlib import json import avro.schema try: unicode except NameError: unicode = str try: basestring # type: ignore except NameError: basestring = (bytes, unicode) # # Constants # # TODO(hammer): confirmed 'fixed' with Doug VALID_TYPE_SCHEMA_TYPES = ('enum', 'record', 'error', 'fixed') # # Exceptions # class ProtocolParseException(avro.schema.AvroException): pass # # Base Classes # class Protocol(object): """An application protocol.""" def _parse_types(self, types, type_names): type_objects = [] for type in types: type_object = avro.schema.make_avsc_object(type, type_names) if type_object.type not in VALID_TYPE_SCHEMA_TYPES: fail_msg = 'Type %s not an enum, fixed, record, or error.' % type raise ProtocolParseException(fail_msg) type_objects.append(type_object) return type_objects def _parse_messages(self, messages, names): message_objects = {} for name, body in messages.items(): if name in message_objects: fail_msg = 'Message name "%s" repeated.' % name raise ProtocolParseException(fail_msg) try: request = body.get('request') response = body.get('response') errors = body.get('errors') except AttributeError: fail_msg = 'Message name "%s" has non-object body %s.' % (name, body) raise ProtocolParseException(fail_msg) message_objects[name] = Message(name, request, response, errors, names) return message_objects def __init__(self, name, namespace=None, types=None, messages=None): # Ensure valid ctor args if not name: fail_msg = 'Protocols must have a non-empty name.' raise ProtocolParseException(fail_msg) elif not isinstance(name, basestring): fail_msg = 'The name property must be a string.' raise ProtocolParseException(fail_msg) elif not (namespace is None or isinstance(namespace, basestring)): fail_msg = 'The namespace property must be a string.' raise ProtocolParseException(fail_msg) elif not (types is None or isinstance(types, list)): fail_msg = 'The types property must be a list.' raise ProtocolParseException(fail_msg) elif not (messages is None or callable(getattr(messages, 'get', None))): fail_msg = 'The messages property must be a JSON object.' raise ProtocolParseException(fail_msg) self._props = {} self.set_prop('name', name) type_names = avro.schema.Names() if namespace is not None: self.set_prop('namespace', namespace) type_names.default_namespace = namespace if types is not None: self.set_prop('types', self._parse_types(types, type_names)) if messages is not None: self.set_prop('messages', self._parse_messages(messages, type_names)) self._md5 = hashlib.md5(str(self).encode()).digest() # read-only properties @property def name(self): return self.get_prop('name') @property def namespace(self): return self.get_prop('namespace') @property def fullname(self): return avro.schema.Name(self.name, self.namespace, None).fullname @property def types(self): return self.get_prop('types') @property def types_dict(self): return {type.name: type for type in self.types} @property def messages(self): return self.get_prop('messages') @property def md5(self): return self._md5 @property def props(self): return self._props # utility functions to manipulate properties dict def get_prop(self, key): return self.props.get(key) def set_prop(self, key, value): self.props[key] = value def to_json(self): to_dump = {} to_dump['protocol'] = self.name names = avro.schema.Names(default_namespace=self.namespace) if self.namespace: to_dump['namespace'] = self.namespace if self.types: to_dump['types'] = [t.to_json(names) for t in self.types] if self.messages: messages_dict = {} for name, body in self.messages.items(): messages_dict[name] = body.to_json(names) to_dump['messages'] = messages_dict return to_dump def __str__(self): return json.dumps(self.to_json()) def __eq__(self, that): to_cmp = json.loads(str(self)) return to_cmp == json.loads(str(that)) class Message(object): """A Protocol message.""" def _parse_request(self, request, names): if not isinstance(request, list): fail_msg = 'Request property not a list: %s' % request raise ProtocolParseException(fail_msg) return avro.schema.RecordSchema(None, None, request, names, 'request') def _parse_response(self, response, names): if isinstance(response, basestring) and names.has_name(response, None): return names.get_name(response, None) else: return avro.schema.make_avsc_object(response, names) def _parse_errors(self, errors, names): if not isinstance(errors, list): fail_msg = 'Errors property not a list: %s' % errors raise ProtocolParseException(fail_msg) errors_for_parsing = {'type': 'error_union', 'declared_errors': errors} return avro.schema.make_avsc_object(errors_for_parsing, names) def __init__(self, name, request, response, errors=None, names=None): self._name = name self._props = {} self.set_prop('request', self._parse_request(request, names)) self.set_prop('response', self._parse_response(response, names)) self.set_prop('errors', self._parse_errors(errors or [], names)) # read-only properties name = property(lambda self: self._name) request = property(lambda self: self.get_prop('request')) response = property(lambda self: self.get_prop('response')) errors = property(lambda self: self.get_prop('errors')) props = property(lambda self: self._props) # utility functions to manipulate properties dict def get_prop(self, key): return self.props.get(key) def set_prop(self, key, value): self.props[key] = value def __str__(self): return json.dumps(self.to_json()) def to_json(self, names=None): if names is None: names = avro.schema.Names() to_dump = {} to_dump['request'] = self.request.to_json(names) to_dump['response'] = self.response.to_json(names) if self.errors: to_dump['errors'] = self.errors.to_json(names) return to_dump def __eq__(self, that): return self.name == that.name and self.props == that.props def make_avpr_object(json_data): """Build Avro Protocol from data parsed out of JSON string.""" try: name = json_data.get('protocol') namespace = json_data.get('namespace') types = json_data.get('types') messages = json_data.get('messages') except AttributeError: raise ProtocolParseException('Not a JSON object: %s' % json_data) return Protocol(name, namespace, types, messages) def parse(json_string): """Constructs the Protocol from the JSON text.""" try: json_data = json.loads(json_string) except ValueError: raise ProtocolParseException('Error parsing JSON: %s' % json_string) # construct the Avro Protocol object return make_avpr_object(json_data)