Protoplasm6

Utilities for working with Protobuf & gRPC in Python: Python 3 dataclasses generated from compiled protobuf messages and casting between the two, quickly initializing large/nested Protobuf objects, and simplified gRPC service interfaces where you implement plain Python methods with plain Python arguments and return values instead of fiddling with request/response messages.

See the Public API index for a task-oriented reference of everything protoplasm6 offers.

Note

This is a fork of the protoplasm package (v5.3.1) by CCP Games, made in order to continue its development and maintenance. The package and module are now called protoplasm6 (starting at version 6.0.0) so it can live side-by-side with the original.

Protoplasm6 & Neobuilder6

Protoplasm6 is the runtime half of a pair. The other half is neobuilder6, the build-time code generator:

                 neobuilder6 (build time)                     protoplasm6 (run time)
 foo.proto ──protoc──► foo_pb2.py, foo_pb2_grpc.py
                    └─► foo_dc.py             (dataclasses) ──► protoplasm6.plasm.DataclassBase
                    └─► foo_api.py            (interfaces)  ──► protoplasm6.plasm errors/decorators
                    └─► foo_grpc_receiver.py  (servicers)   ──► protoplasm6.plasm.BaseGrpcServicer
                    └─► foo_grpc_sender.py    (clients)     ──► protoplasm6.plasm.BaseGrpcClientImplementation

• Neobuilder6 depends on Protoplasm6 (it uses some of its utilities, and the code it generates imports protoplasm6).
• Protoplasm6 does not depend on Neobuilder6 at runtime, but its unit tests use Neobuilder6 to build the test protos (so Neobuilder6 is installed with --no-deps for testing).

The major and minor versions of the two go hand in hand, so code built with Neobuilder6 6.x should be run with Protoplasm6 6.x (where Protoplasm6's minor version is the same or higher).

Installation

pip install protoplasm6

...although you'll most likely want neobuilder6 as well to build your proto files, which pulls in protoplasm6 for you:

pip install neobuilder6

Protoplasm6 requires Python 3.10+ and depends on protobuf 7.x, grpcio and batutils.

Note that *_pb2.py files generated by protoc (i.e. Neobuilder6) check the version of the protobuf runtime when imported, so the protobuf version used to run a generated package must be the same major version and at least the same minor version as the grpcio-tools that generated it.

The idea

Design your whole system of services and messages in proto files, and implement each service as a plain Python class with plain Python methods, arguments and return values. The generated client for a service implements the exact same interface as the service itself, so the code calling a service can't tell (and doesn't care) whether it's talking to a local object in the same process or a remote gRPC service.

That means you can start with a monolith where everything runs in the same runtime, and later pick out any service "wholesale" and move it to its own remote process, swapping the local implementation for the generated gRPC client, without changing the "core code" on either side. No refactoring needed, and nothing in your business logic needs to know what a "Request", "Response", protobuf or even protoplasm is.

Usage

The examples below assume a proto file like this one, at ./proto/myshop/orders/orders.proto, built by Neobuilder6 (see its README):

syntax = "proto3";

package myshop.orders;

import "google/protobuf/timestamp.proto";

service OrderService {
    rpc GetOrder(GetOrderRequest) returns (GetOrderResponse) {}
    rpc ListOrders(ListOrdersRequest) returns (ListOrdersResponse) {}
}

message Customer {
    string customer_id = 1;
    string name = 2;
}

message Order {
    string order_id = 1;
    Customer customer = 2;
    repeated string items = 3;
    google.protobuf.Timestamp placed_at = 4;
}

message GetOrderRequest { string order_id = 1; }
message GetOrderResponse { Order order = 1; }

message ListOrdersRequest { string customer_id = 1; }
message ListOrdersResponse { repeated Order orders = 1; }

That gives you the modules orders_pb2, orders_pb2_grpc, orders_dc, orders_api, orders_grpc_receiver and orders_grpc_sender in the myshop.orders package.

Dataclasses

Every proto message gets a dataclass in the *_dc.py module that extends protoplasm6.plasm.DataclassBase:

import datetime
from myshop.orders.orders_dc import *

order = Order(order_id='o1', customer=Customer(customer_id='c1', name='Jane'),
              items=['apple', 'pear'], placed_at=datetime.datetime.now())

order.to_dict()      # Nested dict of JSON serializable values
order.to_json()
order.to_proto()     # The orders_pb2.Order protobuf message
order.to_protostr()  # Serialized protobuf bytes
order.to_base64()    # Serialized protobuf as base64

Order.from_dict({'order_id': 'o1', 'customer': {'customer_id': 'c1'}})
Order.from_json(...)
Order.from_proto(...)
Order.from_protostr(...)
Order.from_base64(...)
Order.from_kwdict(order_id='o1', customer__name='Jane')  # Double underscores for nesting

Nested messages can be given as dicts, lists/tuples (as positional args) or dataclass instances and will be forced into the correct dataclass type. google.protobuf.Timestamp and Duration fields are datetime.datetime and datetime.timedelta, Any fields hold any dataclass, and Struct/Value fields hold plain dicts/values.

Dataclasses of messages with oneof groups behave like protobuf messages: setting one member resets the others, and you can ask which one is set:

payment.which_oneof('method')  # e.g. 'card_number' or None
payment.get_oneof('method')    # The value of the member that's set (or None)
payment.clear_oneof('method')

Casting utilities

from protoplasm6 import casting
from myshop.orders import orders_pb2

casting.mkproto(orders_pb2.Order, order_id='o1', customer__name='Jane')
casting.mkdataclass(Order, order_id='o1', customer__name='Jane')
casting.proto_to_dataclass(some_proto)
casting.dataclass_to_proto(some_dataclass)
casting.proto_to_dict(some_proto)
casting.dict_to_proto(orders_pb2.Order, {...})
casting.dict_to_dataclass(Order, {...})

Converters (protobuf <-> dataclass)

Whenever protoplasm6 casts between protobuf messages and dataclasses (e.g. to_proto(), from_proto(), casting.proto_to_dataclass(), casting.dataclass_to_proto() and in every gRPC call) it uses the current converter. There are two:

Name	Class	What it does
direct (default)	DirectConverter	Converts directly between messages and dataclasses using the protobuf message API
dict	DictConverter	The original (protoplasm 5.x) conversion: protobuf <-> json_format dict <-> dataclass

The direct converter is roughly 4-7 times faster (both ways, for small and large messages alike) and also gets a few things right that the dict one gets wrong (see the CHANGELOG), e.g. Dict[int, ...] maps keep their int keys, Dict[bool, ...] maps work, timezone aware datetime values are converted to UTC properly and messages with oneof scalar fields work.

Fields it doesn't know how to handle (e.g. ones with your own custom dictators) fall back to the dict logic for that field only.

You can switch converters globally, temporarily or via an environment variable:

from protoplasm6.casting import converters

converters.set_converter('dict')  # Globally

with converters.use_converter('dict'):  # Only within this block (and context/thread)
    proto = order.to_proto()

converters.get_converter()  # The converter currently in use

converters.DictConverter().dataclass_to_proto(order)  # Or use one explicitly

PROTOPLASM6_CONVERTER=dict python my_service.py

You can also roll your own by extending converters.BaseConverter and registering it with converters.register_converter(...).

Services

For every service in a proto file you get a *Interface class in the *_api.py module, with a method per RPC that takes the request message's fields as arguments and returns the response message's field(s), i.e. the request/response messages are "exploded" for you:

from protoplasm6 import plasm
from myshop.orders.orders_api import *
from myshop.orders.orders_dc import *


class MyOrderService(OrderServiceInterface):
    @plasm.require_params  # Raises InvalidArgument if any params are missing
    def get_order(self, order_id: str) -> Order:
        order = my_database.get(order_id)
        if not order:
            raise plasm.NotFound()  # Becomes a NOT_FOUND gRPC status code
        return order

    def list_orders(self, customer_id: str = '') -> list[Order]:
        return my_database.find(customer_id=customer_id)

Unimplemented methods raise plasm.Unimplemented (gRPC UNIMPLEMENTED).

Serving

The interface knows its gRPC servicer class (from *_grpc_receiver.py) so you can do:

from protoplasm6 import plasm

server = plasm.GrpcServer('[::]:50051')
server.add_servicer(OrderServiceInterface.__servicer_cls__(MyOrderService()))
server.serve()

...or add the servicer to your own grpc.server(...) with servicer.add_to_server(grpc_server).

Calling

The *_grpc_sender.py module has a client class (named after the service) that implements the same interface over gRPC:

from myshop.orders.orders_grpc_sender import OrderService

orders: OrderServiceInterface = OrderService('localhost:50051')  # credentials=True for SSL, options={...} for channel options
# ...or, while it's all still one monolith: orders = MyOrderService()

order = orders.get_order('o1')

try:
    orders.get_order('nope')
except plasm.NotFound:
    ...  # gRPC status codes are raised as the corresponding plasm errors

Call metadata (headers)

To send metadata (i.e. headers, e.g. for authorization) with calls, give the client a metadata dict (or a callable returning one, which gets called for every call, e.g. to refresh tokens) and/or use the call_metadata context manager:

orders = OrderService('localhost:50051', metadata={'authorization': f'Bearer {token}'})

with plasm.call_metadata({'x-request-id': 'r1'}):  # For all calls in this block
    orders.get_order('o1')

On the service side, decorate a method with @plasm.takes_context to get a context argument and read headers with context.get_meta('authorization').

Streaming RPCs (stream in the proto) are supported as well, with plasm.RequestIterator / plasm.ResponseIterator on either side (see tests/servicetestutils and the river.proto tests).

Decorators

• plasm.require_params - Raises InvalidArgument if required params are None. Use @plasm.require_params for all params, or @plasm.require_params('foo', 'bar') for specific ones, and typecheck=True to also check the params against their type annotations (both typing style annotations like Optional[List[int]] and modern ones like list[int] | None are supported).
• plasm.takes_context - Marks a method as taking the gRPC call context as a keyword argument (context by default).

Errors

protoplasm6.plasm (and protoplasm6.errors) has an exception for every gRPC status code: NotFound, InvalidArgument, Unimplemented, PermissionDenied, Unauthenticated, Unavailable, DeadlineExceeded, AlreadyExists, FailedPrecondition, DataLoss, ResourceExhausted, OutOfRange, Internal, Cancelled, Aborted and Unknown. Raise them in your implementation and they're raised again on the client side.

Migrating from protoplasm 5.x

1. Replace the protoplasm and neobuilder requirements with protoplasm6 >=6.0, <7 and neobuilder6 >=6.0, <7 (and ccptools with batutils, if you used it).
2. Rebuild your protos with Neobuilder6 (the generated code now imports protoplasm6, and the *_pb2.py files need protobuf 7).
3. Replace protoplasm with protoplasm6 in your own imports, e.g. from protoplasm import plasm -> from protoplasm6 import plasm.
4. Make sure you're on Python 3.10 or later.

The API itself is unchanged.

Development

The unit tests need Neobuilder6 to build the test protos. With both repos checked out side by side:

python -m venv .venv
.venv/Scripts/activate   # or: source .venv/bin/activate
pip install -r requirements-unittesting.txt
pip install --no-deps -e . -e ../neobuilder6
python -m unittest discover -v -f ./tests

CI/CD

GitHub Actions run the unit tests (on Python 3.10 to 3.14) for pushes to main and for all pull requests, and publish to PyPI (after running the tests again and checking that the release tag, e.g. v6.1.0 or 6.1.0, matches __version__) whenever a GitHub release (or pre-release) is created. Publishing needs a PYPI_UPLOAD_TOKEN repository secret.

Since the tests of each package need the other one, CI installs Neobuilder6 (without its dependencies) from the branch with the same name in its repo if there is one (so changes spanning both repos can be tested together, e.g. with a pull request from the same branch name in both) or from main otherwise. When publishing, Neobuilder6 (without its dependencies) comes from PyPI if a matching release is there (otherwise from main).

When releasing both packages, release Protoplasm6 first, then Neobuilder6.