Skip to main content

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)

Release files for proxo 1.0.2

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for proxo 1.0.2
File Size Uploaded
proxo-1.0.2.tar.gz 5.3 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for proxo 1.0.2
File Interpreter ABI Platform
proxo-1.0.2-py2.py3-none-any.whl Python 2, Python 3 none any Details

Total release size: 13.1 kB

Release files / proxo-1.0.2.tar.gz

Download URL proxo-1.0.2.tar.gz
Size 5.3 kB
Tags Source
SHA-256 checksum
How to use checksums
bc8f41669c29eb03d0b36d73c9786f4dd7e21e7699ca39ecaf1f7f06550d5da7
BLAKE2b-256 checksum
How to use checksums
0b8e76b635b583e3660602ffe68e29d6a3e0cf5a53678b57c29f827f96981088
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No

Release files / proxo-1.0.2-py2.py3-none-any.whl

Download URL proxo-1.0.2-py2.py3-none-any.whl
Size 7.7 kB
Tags Python 2 Python 3
SHA-256 checksum
How to use checksums
10f42447526b3098e40e100b82206a5e1e451550ac07d157f7ca11e625994315
BLAKE2b-256 checksum
How to use checksums
c4c00fa608b5ff7d71d4b0455d079fe44599e0feaf711232a98a0648cb2e9d69
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No

Release history Release notifications | RSS feed

This release

1.0.2 This release

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page