Skip to main content

Object proxies (wrappers) for protobuf messages

Project description

Build Status

Proxo

Extend protobuf message with custom methods properties and additional attributes

TL;DR

from proxo import MessageProxy, encode, decode

class Person(MessageProxy):
    proto = addressbook_pb2.Person  # it can be more complex, like pattern matching, see below

    @property
    def firstname(self):
        return self.name.split(' ')[0]


p = Person(name='Test Me')
assert p.firstname == 'Test'
assert decode(encode(p)) == p

Usage

Given the addressbook protobuf definition

package tutorial;

message Person {
  required string name = 1;
  required int32 id = 2;
  optional string email = 3;

  enum PhoneType {
    MOBILE = 0;
    HOME = 1;
    WORK = 2;
  }

  message PhoneNumber {
    required string number = 1;
    optional PhoneType type = 2 [default = HOME];
  }

  repeated PhoneNumber phone = 4;
}

message AddressBook {
  repeated Person person = 1;
}

The traditional way

import addressbook_pb2

person = addressbook_pb2.Person()
person.id = 1234
person.name = "John Doe"
person.email = "jdoe@example.com"
phone = person.phone.add()
phone.number = "555-4321"
phone.type = addressbook_pb2.Person.HOME

via Proxo.dict_to_protobuf

from proxo import dict_to_protobuf, protobuf_to_dict

data = {'id': 124,
        'name': 'John Doe',
        'email': 'jdoe@example.com',
        'phone': {'number': '555-4321',
                  'type': 'HOME'}}

proto = dict_to_protobuf(data, addressbook_pb2.Person)

assert person == proto

# converting back
mapping = protobuf_to_dict(proto)
mapping['phone']['number']
mapping.phone.number  # using dot notation

assert mapping == data

via extending Proxo.MessageProxy

from proxo import MessageProxy, encode, decode

# note that non defined types will be automatically proxied too

class Person(MessageProxy):
    proto = addressbook_pb2.Person  # it can be more complex, like pattern matching, see below

    @property
    def firstname(self):
        return self.name.split(' ')[0]

    def call(self):
        try:
            print('calling {}'.format(self.firstname))
            do_voip_call(self.phone.number)
        except:
            print('failed calling {} on his/her {} number'.format(self.firstname,
                                                                  self.phone.type.lower))

obj = Person(id=124, name='John Doe', phone={'number': '555-4321',
                                             'type': 'HOME'})
obj.phone.type = 'MOBILE'
assert obj.firsname == 'John'

proto = encode(obj)
john = decode(proto)

# lets bother him
john.call()

More Complicated Example

import operator

from uuid import uuid4
from functools import partial
from proxo import MessageProxy


class Scalar(MessageProxy):
    proto = mesos_pb2.Value.Scalar


class Resource(MessageProxy):
    proto = mesos_pb2.Resource  # can be class


class ScalarResource(Resource):
    proto = mesos_pb2.Resource(type=mesos_pb2.Value.SCALAR)  # or partially set instance

    def __init__(self, value=None, **kwargs):
        super(Resource, self).__init__(**kwargs)
        if value is not None:
            self.scalar = Scalar(value=value)

    def __cmp__(self, other):
        first, second = float(self), float(other)
        if first < second:
            return -1
        elif first > second:
            return 1
        else:
            return 0

    def __repr__(self):
        return "<{}: {}>".format(self.__class__.__name__, self.scalar.value)

    def __float__(self):
        return float(self.scalar.value)

    @classmethod
    def _op(cls, op, first, second):
        value = op(float(first), float(second))
        return cls(value=value)

    def __add__(self, other):
        return self._op(operator.add, self, other)

    def __radd__(self, other):
        return self._op(operator.add, other, self)

    def __sub__(self, other):
        return self._op(operator.sub, self, other)

    def __rsub__(self, other):
        return self._op(operator.sub, other, self)

    def __mul__(self, other):
        return self._op(operator.mul, self, other)

    def __rmul__(self, other):
        return self._op(operator.mul, other, self)

    def __truediv__(self, other):
        return self._op(operator.truediv, self, other)

    def __rtruediv__(self, other):
        return self._op(operator.truediv, other, self)

    def __iadd__(self, other):
        self.scalar.value = float(self._op(operator.add, self, other))
        return self

    def __isub__(self, other):
        self.scalar.value = float(self._op(operator.sub, self, other))
        return self


class Cpus(ScalarResource):
    proto = mesos_pb2.Resource(name='cpus', type=mesos_pb2.Value.SCALAR)


class Mem(ScalarResource):
    proto = mesos_pb2.Resource(name='mem', type=mesos_pb2.Value.SCALAR)


class Disk(ScalarResource):
    proto = mesos_pb2.Resource(name='disk', type=mesos_pb2.Value.SCALAR)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

proxo-1.0.2.tar.gz (5.3 kB view details)

Uploaded Source

Built Distribution

proxo-1.0.2-py2.py3-none-any.whl (7.7 kB view details)

Uploaded Python 2 Python 3

File details

Details for the file proxo-1.0.2.tar.gz.

File metadata

  • Download URL: proxo-1.0.2.tar.gz
  • Upload date:
  • Size: 5.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No

File hashes

Hashes for proxo-1.0.2.tar.gz
Algorithm Hash digest
SHA256 bc8f41669c29eb03d0b36d73c9786f4dd7e21e7699ca39ecaf1f7f06550d5da7
MD5 def6fe7e2c2f9e48ba493196002aefaf
BLAKE2b-256 0b8e76b635b583e3660602ffe68e29d6a3e0cf5a53678b57c29f827f96981088

See more details on using hashes here.

File details

Details for the file proxo-1.0.2-py2.py3-none-any.whl.

File metadata

File hashes

Hashes for proxo-1.0.2-py2.py3-none-any.whl
Algorithm Hash digest
SHA256 10f42447526b3098e40e100b82206a5e1e451550ac07d157f7ca11e625994315
MD5 8cbbd6665be62136a26c1127241c21a3
BLAKE2b-256 c4c00fa608b5ff7d71d4b0455d079fe44599e0feaf711232a98a0648cb2e9d69

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page