Sérialisation avec protobuf

protobuf optimise la sérialisation de deux façons. Elle accélère l’écriture et la lecture des données et permet aussi un accès rapide à une information précise dans désérialiser les autres. Elle réalise cela en imposant un schéma strict de données.

L’exemple fonctionne si l’exécutable protoc et le package protobuf ont des versions compatibles. Un message apparaîtra dans le cas contraire.

protoc --version
python -c "import google.protobuf as gp;print(gp.__version__)"

Schéma

On récupère l’exemple du tutorial.

import os
import sys
import timeit
import struct
from io import BytesIO
from sphinx_runpython.runpython import run_cmd
import google.protobuf as gp
from google.protobuf.json_format import MessageToJson, Parse as ParseJson

schema = """
syntax = "proto2";

package tutorial;

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

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

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

  repeated PhoneNumber phones = 4;
}

message AddressBook {
  repeated Person people = 1;
}
"""

Compilation

Il faut d’abord récupérer le compilateur. Cela peut se faire depuis le site de protobuf ou sur Linux (Ubuntu/Debian) apt-get install protobuf-compiler pour obtenir le programme protoc.

version = gp.__version__
version

'4.23.4'

with open("schema.proto", "w") as f:
    f.write(schema)


# Et on peut compiler.

# In[8]:


cmd = "protoc --python_out=. schema.proto"
try:
    out, err = run_cmd(cmd=cmd, wait=True)
    use_protoc = True
except FileNotFoundError as e:
    print(f"error: {e}")
    print("unable to use protoc")
    use_protoc = False
if use_protoc:
    print(out)
    print(err)

error: [Errno 2] No such file or directory: 'protoc'
unable to use protoc

Un fichier a été généré.

[_ for _ in os.listdir(".") if ".py" in _]

['plot_lambda_function.py', 'plot_matador.py', 'plot_pandas_groupby.py', 'plot_tarabiscote.py', 'plot_serialisation_examples.py', 'plot_partie_dame.py', 'plot_serialisation_protobuf.py', 'plot_gil_example.py', 'plot_numpy_tricks.py', 'plot_float_and_double_rouding.py', 'plot_tsp.py', 'plot_hypercube.py', 'plot_einstein_riddle.py']

if os.path.exists("schema_pb2.py"):
    with open("schema_pb2.py", "r") as f:
        content = f.read()
    print(content[:1000])
else:
    print("schema_pb2.py missing")

schema_pb2.py missing

Import du module créé

Pour utliser protobuf, il faut importer le module créé.

if use_protoc:
    sys.path.append(".")
    import schema_pb2  # noqa: E402

On créé un enregistrement.

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

if use_protoc:
    person

Sérialisation en chaîne de caractères

if use_protoc:
    res = person.SerializeToString()
    print(type(res), res)

if use_protoc:
    print(timeit.timeit("person.SerializeToString()", globals=globals(), number=100))

if use_protoc:
    pers = schema_pb2.Person.FromString(res)
    print(pers)

if use_protoc:
    pers = schema_pb2.Person()
    pers.ParseFromString(res)
    print(pers)

if use_protoc:
    print(
        timeit.timeit(
            "schema_pb2.Person.FromString(res)", globals=globals(), number=100
        )
    )

if use_protoc:
    print(timeit.timeit("pers.ParseFromString(res)", globals=globals(), number=100))

Plusieurs chaînes de caractères

db = []
if use_protoc:
    person = schema_pb2.Person()
    person.id = 1234
    person.name = "John Doe"
    person.email = "jdoe@example.com"
    phone = person.phones.add()
    phone.number = "555-4321"
    phone.type = schema_pb2.Person.HOME
    db.append(person)

    person = schema_pb2.Person()
    person.id = 5678
    person.name = "Johnette Doette"
    person.email = "jtdoet@example2.com"
    phone = person.phones.add()
    phone.number = "777-1234"
    phone.type = schema_pb2.Person.MOBILE
    db.append(person)

buffer = BytesIO()
for p in db:
    size = p.ByteSize()
    buffer.write(struct.pack("i", size))
    buffer.write(p.SerializeToString())
res = buffer.getvalue()
res

b''

db2 = []
buffer = BytesIO(res)
n = 0
while True:
    bsize = buffer.read(4)
    if len(bsize) == 0:
        # C'est fini.
        break
    size = struct.unpack("i", bsize)[0]
    data = buffer.read(size)
    p = schema_pb2.Person.FromString(data)
    db2.append(p)

if db2:
    print(db2[0], db2[1])

Sérialisation JSON

if use_protoc:
    print(MessageToJson(pers))

if use_protoc:
    print(timeit.timeit("MessageToJson(pers)", globals=globals(), number=100))

if use_protoc:
    js = MessageToJson(pers)
    res = ParseJson(js, message=schema_pb2.Person())
    print(res)

if use_protoc:
    print(
        timeit.timeit(
            "ParseJson(js, message=schema_pb2.Person())", globals=globals(), number=100
        )
    )