CBOR bindings #559
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,226 @@ | ||
| # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| # SPDX-License-Identifier: Apache-2.0. | ||
|
|
||
| import _awscrt | ||
|
|
||
| from awscrt import NativeResource | ||
| from enum import IntEnum | ||
| from typing import Union, Any | ||
|
|
||
|
|
||
| class AwsCborType(IntEnum): | ||
| # Corresponding to `enum aws_cbor_type` in aws/common/cbor.h | ||
| Unknown = 0 | ||
| UnsignedInt = 1 | ||
| NegativeInt = 2 | ||
| Float = 3 | ||
| Bytes = 4 | ||
| Text = 5 | ||
| ArrayStart = 6 | ||
| MapStart = 7 | ||
| Tag = 8 | ||
| Bool = 9 | ||
| Null = 10 | ||
| Undefined = 11 | ||
| Break = 12 | ||
| IndefBytes = 13 | ||
| IndefStr = 14 | ||
| IndefArray = 15 | ||
| IndefMap = 16 | ||
|
|
||
|
|
||
| class AwsCborEncoder(NativeResource): | ||
| """ Encoder for CBOR """ | ||
|
|
||
| def __init__(self): | ||
| super().__init__() | ||
| self._binding = _awscrt.cbor_encoder_new(self) | ||
|
|
||
| def get_encoded_data(self) -> bytes: | ||
| """Return the current encoded data as bytes | ||
|
|
||
| Returns: | ||
| bytes: The encoded data currently | ||
| """ | ||
| return _awscrt.cbor_encoder_get_encoded_data(self._binding) | ||
|
|
||
| def write_int(self, val: int): | ||
| # TODO: maybe not support bignum for now. Not needed? | ||
| """Write an int as cbor formatted, | ||
| val less than -2^64 will be encoded as Negative bignum for CBOR | ||
| val between -2^64 to -1, inclusive, will be encode as negative integer for CBOR | ||
| val between 0 to 2^64 - 1, inclusive, will be encoded as unsigned integer for CBOR | ||
| val greater than 2^64 - 1 will be encoded as Unsigned bignum for CBOR | ||
|
|
||
| Args: | ||
| val (int): value to be encoded and written to the encoded data. | ||
| """ | ||
| assert isinstance(val, int) | ||
|
There was a problem hiding this comment. for my education, havent we already indicated in function signature htat val is an int? why do we have to check twice |
||
| val_to_encode = val | ||
| if val < 0: | ||
| # For negative value, the value to encode is -1 - val. | ||
| val_to_encode = -1 - val | ||
| bit_len = val_to_encode.bit_length() | ||
| if bit_len > 64: | ||
|
There was a problem hiding this comment. wouldnt python need 65 bits to represent - u64_max? so this logic might be a bit off, since we dont need bignum for - u64_max |
||
| # Bignum | ||
| bytes_len = bit_len // 8 | ||
| if bit_len % 8 > 0: | ||
| bytes_len += 1 | ||
| bytes_val = val_to_encode.to_bytes(bytes_len, "big") | ||
| if val < 0: | ||
| self.write_tag(AwsCborTags.NegativeBigNum) # tag for negative bignum | ||
| else: | ||
| self.write_tag(AwsCborTags.UnsignedBigNum) # tag for unsigned bignum | ||
| return self.write_bytes(bytes_val) | ||
|
|
||
| if val >= 0: | ||
| return _awscrt.cbor_encoder_write_unsigned_int(self._binding, val_to_encode) | ||
| else: | ||
| return _awscrt.cbor_encoder_write_negative_int(self._binding, val_to_encode) | ||
|
|
||
| def write_float(self, val: float): | ||
| """Write a double as cbor formatted | ||
| If the val can be convert the int without loss of precision, | ||
| it will be converted to int to be written to as cbor formatted. | ||
|
|
||
| Args: | ||
| val (float): value to be encoded and written to the encoded data. | ||
| """ | ||
| assert isinstance(val, float) | ||
| # Floating point numbers are usually implemented using double in C | ||
| return _awscrt.cbor_encoder_write_float(self._binding, val) | ||
|
|
||
| def write_bytes(self, val: bytes): | ||
| """Write bytes as cbor formatted | ||
|
|
||
| Args: | ||
| val (bytes): value to be encoded and written to the encoded data. | ||
| """ | ||
| return _awscrt.cbor_encoder_write_bytes(self._binding, val) | ||
|
|
||
| def write_text(self, val: str): | ||
| """Write text as cbor formatted | ||
|
|
||
| Args: | ||
| val (str): value to be encoded and written to the encoded data. | ||
| """ | ||
| return _awscrt.cbor_encoder_write_text(self._binding, val) | ||
|
|
||
| def write_array_start(self, number_entries: int): | ||
| """Add a start of array element. | ||
| A legistic with the `number_entries` | ||
|
|
||
| for the cbor data items to be included in the array. | ||
| `number_entries` should 0 to 2^64 inclusive. | ||
| Otherwise, overflow will be raised. | ||
|
|
||
| Args: | ||
| number_entries (int): number of entries in the array to be written | ||
| """ | ||
| if number_entries < 0 or number_entries > 2**64: | ||
|
There was a problem hiding this comment. ugh, cbor spec allows 2^64 entries? not sure we would be able to allocate cbor chunk for that even if all entries are 1 byte. should we consider a lower bound |
||
| raise OverflowError() | ||
|
|
||
| return _awscrt.cbor_encoder_write_array_start(self._binding, number_entries) | ||
|
|
||
| def write_map_start(self, number_entries: int): | ||
| """Add a start of map element, with the `number_entries` | ||
| for the number of pair of cbor data items to be included in the map. | ||
| `number_entries` should 0 to 2^64 inclusive. | ||
| Otherwise, overflow will be raised. | ||
|
|
||
| Args: | ||
| number_entries (int): number of entries in the map to be written | ||
| """ | ||
| if number_entries < 0 or number_entries > 2**64: | ||
| raise ValueError() | ||
|
|
||
| return _awscrt.cbor_encoder_write_map_start(self._binding, number_entries) | ||
|
|
||
| def write_tag(self, tag_number: int): | ||
| if tag_number < 0 or tag_number > 2**64: | ||
| raise ValueError() | ||
|
|
||
| return _awscrt.cbor_encoder_write_tag(self._binding, tag_number) | ||
|
|
||
| def write_null(self): | ||
| return _awscrt.cbor_encoder_write_simple_types(self._binding, AwsCborType.Null) | ||
|
|
||
| def write_bool(self, val: bool): | ||
| return _awscrt.cbor_encoder_write_bool(self._binding, val) | ||
|
|
||
| def write_list(self, val: list): | ||
| return _awscrt.cbor_encoder_write_py_list(self._binding, val) | ||
|
|
||
| def write_dict(self, val: dict): | ||
| return _awscrt.cbor_encoder_write_py_dict(self._binding, val) | ||
|
|
||
| def write_data_item(self, data_item: Any): | ||
| """Generic API to write any type of an data_item as cbor formatted. | ||
| TODO: timestamp <-> datetime?? Decimal fraction <-> decimal?? | ||
|
|
||
| Args: | ||
| data_item (Any): any type of data_item. If the type is not supported to be converted to cbor format, ValueError will be raised. | ||
| """ | ||
| return _awscrt.cbor_encoder_write_data_item(self._binding, data_item) | ||
|
|
||
|
|
||
| class AwsCborDecoder(NativeResource): | ||
| """ Decoder for CBOR """ | ||
|
|
||
| def __init__(self, src: bytes): | ||
| super().__init__() | ||
| self._src = src | ||
| self._binding = _awscrt.cbor_decoder_new(src) | ||
|
|
||
| def peek_next_type(self) -> AwsCborType: | ||
| """Return the AwsCborType of the next data item in the cbor formatted data | ||
| """ | ||
| return AwsCborType(_awscrt.cbor_decoder_peek_type(self._binding)) | ||
|
|
||
| def get_remaining_bytes_len(self) -> int: | ||
| return _awscrt.cbor_decoder_get_remaining_bytes_len(self._binding) | ||
|
|
||
| def consume_next_element(self): | ||
| return _awscrt.cbor_decoder_consume_next_element(self._binding) | ||
|
|
||
| def consume_next_data_item(self): | ||
| return _awscrt.cbor_decoder_consume_next_data_item(self._binding) | ||
|
|
||
| def pop_next_unsigned_int(self) -> int: | ||
| return _awscrt.cbor_decoder_pop_next_unsigned_int(self._binding) | ||
|
|
||
| def pop_next_negative_int(self) -> int: | ||
| val = _awscrt.cbor_decoder_pop_next_negative_int(self._binding) | ||
| return -1 - val | ||
|
|
||
| def pop_next_double(self) -> float: | ||
| return _awscrt.cbor_decoder_pop_next_float(self._binding) | ||
|
|
||
| def pop_next_bool(self) -> bool: | ||
| return _awscrt.cbor_decoder_pop_next_bool(self._binding) | ||
|
|
||
| def pop_next_bytes(self) -> bytes: | ||
| return _awscrt.cbor_decoder_pop_next_bytes(self._binding) | ||
|
|
||
| def pop_next_text(self) -> str: | ||
| return _awscrt.cbor_decoder_pop_next_text(self._binding) | ||
|
|
||
| def pop_next_array_start(self) -> int: | ||
| return _awscrt.cbor_decoder_pop_next_array_start(self._binding) | ||
|
|
||
| def pop_next_map_start(self) -> int: | ||
| return _awscrt.cbor_decoder_pop_next_map_start(self._binding) | ||
|
|
||
| def pop_next_tag_val(self) -> int: | ||
| return _awscrt.cbor_decoder_pop_next_tag_val(self._binding) | ||
|
|
||
| def pop_next_list(self) -> list: | ||
| return _awscrt.cbor_decoder_pop_next_py_list(self._binding) | ||
|
|
||
| def pop_next_map(self) -> dict: | ||
| return _awscrt.cbor_decoder_pop_next_py_dict(self._binding) | ||
|
|
||
| def pop_next_data_item(self) -> Any: | ||
| """Generic API to decode cbor formatted data to a python object. | ||
| TODO: tags are NOT supported yet. | ||
| """ | ||
| return _awscrt.cbor_decoder_pop_next_data_item(self._binding) | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
smithy currently does not support anything above 64bits afaik so we can probably skip bignum for now