danielgtaylor / danielgtaylor/python-betterproto
Add support for "native" JSON / Map that consists of number keys instead of snake case keys
- Dominant language
- Python
- Stars
- 1.8k
- Forks
- 234
- PR merge metrics
- No merged PRs in 30d
Description
It would be nice to have support for deserialization and serialization using the "native" JSON / Map format, where the keys are numbers instead of snake case field names. This is a must to enable interoperability with other clients, eg. Dart.
Here is a simple patch until it is supported by default:
The only lines that were modified are `betterproto.Message.to_dict = to_dict_patch
betterproto.Message.from_dict = from_dict_patch` and `cased_name = str(meta.number)`.
```Python
def to_dict_patch(
self, casing: Casing = Casing.CAMEL, include_default_values: bool = False
) -> dict:
"""
Returns a dict representation of this message instance which can be
used to serialize to e.g. JSON. Defaults to camel casing for
compatibility but can be set to other modes.
`include_default_values` can be set to `True` to include default
values of fields. E.g. an `int32` type field with `0` value will
not be in returned dict if `include_default_values` is set to
`False`.
"""
output: Dict[str, Any] = {}
for field in dataclasses.fields(self):
meta = FieldMetadata.get(field)
v = getattr(self, field.name)
cased_name = str(meta.number)
if meta.proto_type == "message":
if isinstance(v, datetime):
if v != DATETIME_ZERO or include_default_values:
output[cased_name] = _Timestamp.timestamp_to_json(v)
elif isinstance(v, timedelta):
if v != timedelta(0) or include_default_values:
output[cased_name] = _Duration.delta_to_json(v)
elif meta.wraps:
if v is not None or include_default_values:
output[cased_name] = v
elif isinstance(v, list):
# Convert each item.
v = [i.to_dict(casing, include_default_values) for i in v]
if v or include_default_values:
output[cased_name] = v
else:
if v._serialized_on_wire or include_default_values:
output[cased_name] = v.to_dict(casing, include_default_values)
elif meta.proto_type == "map":
for k in v:
if hasattr(v[k], "to_dict"):
v[k] = v[k].to_dict(casing, include_default_values)
if v or include_default_values:
output[cased_name] = [{'1': k, '2': v} for k, v in v.items()]
elif v != self._get_field_default(field, meta) or include_default_values:
if meta.proto_type in INT_64_TYPES:
if isinstance(v, list):
output[cased_name] = [str(n) for n in v]
else:
output[cased_name] = str(v)
elif meta.proto_type == TYPE_BYTES:
if isinstance(v, list):
output[cased_name] = [b64encode(b).decode("utf8") for b in v]
else:
output[cased_name] = b64encode(v).decode("utf8")
elif meta.proto_type == TYPE_ENUM:
enum_values = list(
self._betterproto.cls_by_field[field.name]
) # type: ignore
if isinstance(v, list):
output[cased_name] = [enum_values[e].name for e in v]
else:
output[cased_name] = enum_values[v].name
else:
output[cased_name] = v
return output
def from_dict_patch(self: T, value: dict) -> T:
"""
Parse the key/value pairs in `value` into this message instance. This
returns the instance itself and is therefore assignable and chainable.
"""
self._serialized_on_wire = True
fields_by_name = {str(FieldMetadata.get(f).number): f for f in dataclasses.fields(self)}
for key in value:
snake_cased = safe_snake_case(key)
if snake_cased in fields_by_name:
field = fields_by_name[snake_cased]
meta = FieldMetadata.get(field)
if value[key] is not None:
if meta.proto_type == "message":
v = getattr(self, field.name)
if isinstance(v, list):
cls = self._betterproto.cls_by_field[field.name]
for i in range(len(value[key])):
v.append(cls().from_dict(value[key][i]))
elif isinstance(v, datetime):
v = datetime.fromisoformat(
value[key].replace("Z", "+00:00")
)
setattr(self, field.name, v)
elif isinstance(v, timedelta):
v = timedelta(seconds=float(value[key][:-1]))
setattr(self, field.name, v)
elif meta.wraps:
setattr(self, field.name, value[key])
else:
v.from_dict(value[key])
elif meta.map_types and meta.map_types[1] == TYPE_MESSAGE:
v = getattr(self, field.name)
cls = self._betterproto.cls_by_field[field.name + ".value"]
for k in value[key]:
v[k] = cls().from_dict(value[key][k])
else:
v = value[key]
if meta.proto_type in INT_64_TYPES:
if isinstance(value[key], list):
v = [int(n) for n in value[key]]
else:
v = int(value[key])
elif meta.proto_type == TYPE_BYTES:
if isinstance(value[key], list):
v = [b64decode(n) for n in value[key]]
else:
v = b64decode(value[key])
elif meta.proto_type == TYPE_ENUM:
enum_cls = self._betterproto.cls_by_field[field.name]
if isinstance(v, list):
v = [enum_cls.from_string(e) for e in v]
elif isinstance(v, str):
v = enum_cls.from_string(v)
if v is not None:
setattr(self, field.name, v)
return self
betterproto.Message.to_dict = to_dict_patch
betterproto.Message.from_dict = from_dict_patch
```
Contributor guide
Assessment
This issue has not been assessed yet.