lists.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  1. """
  2. Module for sedes objects that use lists as serialization format.
  3. """
  4. from collections.abc import (
  5. Sequence,
  6. )
  7. from eth_utils import (
  8. to_list,
  9. to_tuple,
  10. )
  11. from rlp.exceptions import (
  12. DeserializationError,
  13. ListDeserializationError,
  14. ListSerializationError,
  15. SerializationError,
  16. )
  17. from .binary import (
  18. Binary as BinaryClass,
  19. )
  20. def is_sedes(obj):
  21. """
  22. Check if `obj` is a sedes object.
  23. A sedes object is characterized by having the methods `serialize(obj)` and
  24. `deserialize(serial)`.
  25. """
  26. return hasattr(obj, "serialize") and hasattr(obj, "deserialize")
  27. def is_sequence(obj):
  28. """Check if `obj` is a sequence, but not a string or bytes."""
  29. return isinstance(obj, Sequence) and not (
  30. isinstance(obj, str) or BinaryClass.is_valid_type(obj)
  31. )
  32. class List(list):
  33. """
  34. A sedes for lists, implemented as a list of other sedes objects.
  35. :param strict: If true (de)serializing lists that have a length not
  36. matching the sedes length will result in an error. If false
  37. (de)serialization will stop as soon as either one of the
  38. lists runs out of elements.
  39. """
  40. def __init__(self, elements=None, strict=True):
  41. super().__init__()
  42. self.strict = strict
  43. if elements:
  44. for e in elements:
  45. if is_sedes(e):
  46. self.append(e)
  47. elif isinstance(e, Sequence):
  48. self.append(List(e))
  49. else:
  50. raise TypeError(
  51. "Instances of List must only contain sedes objects or "
  52. "nested sequences thereof."
  53. )
  54. @to_list
  55. def serialize(self, obj):
  56. if not is_sequence(obj):
  57. raise ListSerializationError("Can only serialize sequences", obj)
  58. if self.strict and len(self) != len(obj):
  59. raise ListSerializationError(
  60. "Serializing list length (%d) does not match sedes (%d)"
  61. % (len(obj), len(self)),
  62. obj,
  63. )
  64. for index, (element, sedes) in enumerate(zip(obj, self)):
  65. try:
  66. yield sedes.serialize(element)
  67. except SerializationError as e:
  68. raise ListSerializationError(obj=obj, element_exception=e, index=index)
  69. @to_tuple
  70. def deserialize(self, serial):
  71. if not is_sequence(serial):
  72. raise ListDeserializationError("Can only deserialize sequences", serial)
  73. if self.strict and len(serial) != len(self):
  74. raise ListDeserializationError(
  75. "Deserializing list length (%d) does not match sedes (%d)"
  76. % (len(serial), len(self)),
  77. serial,
  78. )
  79. for idx, (sedes, element) in enumerate(zip(self, serial)):
  80. try:
  81. yield sedes.deserialize(element)
  82. except DeserializationError as e:
  83. raise ListDeserializationError(
  84. serial=serial, element_exception=e, index=idx
  85. )
  86. class CountableList:
  87. """
  88. A sedes for lists of arbitrary length.
  89. :param element_sedes: when (de-)serializing a list, this sedes will be
  90. applied to all of its elements
  91. :param max_length: maximum number of allowed elements, or `None` for no limit
  92. """
  93. def __init__(self, element_sedes, max_length=None):
  94. self.element_sedes = element_sedes
  95. self.max_length = max_length
  96. @to_list
  97. def serialize(self, obj):
  98. if not is_sequence(obj):
  99. raise ListSerializationError("Can only serialize sequences", obj)
  100. if self.max_length is not None and len(obj) > self.max_length:
  101. raise ListSerializationError(
  102. f"Too many elements ({len(obj)}, allowed {self.max_length})",
  103. obj=obj,
  104. )
  105. for index, element in enumerate(obj):
  106. try:
  107. yield self.element_sedes.serialize(element)
  108. except SerializationError as e:
  109. raise ListSerializationError(obj=obj, element_exception=e, index=index)
  110. @to_tuple
  111. def deserialize(self, serial):
  112. if not is_sequence(serial):
  113. raise ListDeserializationError(
  114. "Can only deserialize sequences", serial=serial
  115. )
  116. for index, element in enumerate(serial):
  117. if self.max_length is not None and index >= self.max_length:
  118. raise ListDeserializationError(
  119. f"Too many elements (more than {self.max_length})",
  120. serial=serial,
  121. )
  122. try:
  123. yield self.element_sedes.deserialize(element)
  124. except DeserializationError as e:
  125. raise ListDeserializationError(
  126. serial=serial, element_exception=e, index=index
  127. )