1010from datetime import datetime
1111from decimal import Decimal
1212from functools import wraps
13- from inspect import isclass
13+ from inspect import getfullargspec , isclass
1414from typing import (
1515 Any ,
1616 Callable ,
1717 ClassVar ,
1818 Dict ,
19+ Generic ,
20+ Iterable ,
1921 List ,
2022 Optional ,
23+ Set ,
2124 Type ,
2225 TypeVar ,
2326 Union ,
27+ cast ,
2428 get_type_hints ,
2529)
2630
31+ import cbor2
32+
33+ from pycardano .logging import logger
34+
35+ # Remove the semantic decoder for 258 (CBOR tag for set) as we care about the order of elements
36+ try :
37+ cbor2 ._decoder .semantic_decoders .pop (258 )
38+ except Exception as e :
39+ logger .warning ("Failed to remove semantic decoder for CBOR tag 258" , e )
40+ pass
41+
2742from cbor2 import CBOREncoder , CBORSimpleValue , CBORTag , dumps , loads , undefined
2843from frozendict import frozendict
2944from frozenlist import FrozenList
4459 "RawCBOR" ,
4560 "list_hook" ,
4661 "limit_primitive_type" ,
62+ "OrderedSet" ,
63+ "NonEmptyOrderedSet" ,
4764]
4865
66+ T = TypeVar ("T" )
67+
4968
5069def _identity (x ):
5170 return x
@@ -314,10 +333,12 @@ def validate(self):
314333 def _check_recursive (value , type_hint ):
315334 if type_hint is Any :
316335 return True
336+
337+ if isinstance (value , CBORSerializable ):
338+ value .validate ()
339+
317340 origin = getattr (type_hint , "__origin__" , None )
318341 if origin is None :
319- if isinstance (value , CBORSerializable ):
320- value .validate ()
321342 return isinstance (value , type_hint )
322343 elif origin is ClassVar :
323344 return _check_recursive (value , type_hint .__args__ [0 ])
@@ -329,7 +350,7 @@ def _check_recursive(value, type_hint):
329350 _check_recursive (k , key_type ) and _check_recursive (v , value_type )
330351 for k , v in value .items ()
331352 )
332- elif origin in (list , set , tuple , frozenset ):
353+ elif origin in (list , set , tuple , frozenset , OrderedSet ):
333354 if value is None :
334355 return True
335356 args = type_hint .__args__
@@ -364,12 +385,15 @@ def to_validated_primitive(self) -> Primitive:
364385 return self .to_primitive ()
365386
366387 @classmethod
367- def from_primitive (cls : Type [CBORBase ], value : Any ) -> CBORBase :
388+ def from_primitive (
389+ cls : Type [CBORBase ], value : Any , type_args : Optional [tuple ] = None
390+ ) -> CBORBase :
368391 """Turn a CBOR primitive to its original class type.
369392
370393 Args:
371394 cls (CBORBase): The original class type.
372395 value (:const:`Primitive`): A CBOR primitive.
396+ type_args (Optional[tuple]): Type arguments for the class.
373397
374398 Returns:
375399 CBORBase: A CBOR serializable object.
@@ -519,10 +543,26 @@ def _restore_typed_primitive(
519543 Union[:const:`Primitive`, CBORSerializable]: A CBOR primitive or a CBORSerializable.
520544 """
521545
546+ is_cbor_serializable = False
547+ try :
548+ is_cbor_serializable = issubclass (t , CBORSerializable )
549+ except TypeError :
550+ # Handle the case when t is a generic alias
551+ origin = typing .get_origin (t )
552+ if origin is not None :
553+ try :
554+ is_cbor_serializable = issubclass (origin , CBORSerializable )
555+ except TypeError :
556+ pass
557+
522558 if t is Any or (t in PRIMITIVE_TYPES and isinstance (v , t )):
523559 return v
524- elif isclass (t ) and issubclass (t , CBORSerializable ):
525- return t .from_primitive (v )
560+ elif is_cbor_serializable :
561+ if "type_args" in getfullargspec (t .from_primitive ).args :
562+ args = typing .get_args (t )
563+ return t .from_primitive (v , type_args = args )
564+ else :
565+ return t .from_primitive (v )
526566 elif hasattr (t , "__origin__" ) and (t .__origin__ is list ):
527567 t_args = t .__args__
528568 if len (t_args ) != 1 :
@@ -941,3 +981,82 @@ def list_hook(
941981 CBORSerializables.
942982 """
943983 return lambda vals : [cls .from_primitive (v ) for v in vals ]
984+
985+
986+ class OrderedSet (list , Generic [T ], CBORSerializable ):
987+ def __init__ (self , iterable : Optional [List [T ]] = None , use_tag : bool = True ):
988+ super ().__init__ ()
989+ self ._set : Set [str ] = set ()
990+ self ._use_tag = use_tag
991+ if iterable :
992+ self .extend (iterable )
993+
994+ def append (self , item : T ) -> None :
995+ item_key = str (item )
996+ if item_key not in self ._set :
997+ super ().append (item )
998+ self ._set .add (item_key )
999+
1000+ def extend (self , items : Iterable [T ]) -> None :
1001+ for item in items :
1002+ self .append (item )
1003+
1004+ def __contains__ (self , item : object ) -> bool :
1005+ return str (item ) in self ._set
1006+
1007+ def __eq__ (self , other : object ) -> bool :
1008+ if not isinstance (other , OrderedSet ):
1009+ if isinstance (other , list ):
1010+ return list (self ) == other
1011+ return False
1012+ return list (self ) == list (other )
1013+
1014+ def __repr__ (self ) -> str :
1015+ return f"{ self .__class__ .__name__ } { list (self )}
1016+
1017+ def to_shallow_primitive (self ) -> Union [CBORTag , List [T ]]:
1018+ if self ._use_tag :
1019+ return CBORTag (258 , list (self ))
1020+ return list (self )
1021+
1022+ @classmethod
1023+ def from_primitive (
1024+ cls : Type [OrderedSet [T ]], value : Primitive , type_args : Optional [tuple ] = None
1025+ ) -> OrderedSet [T ]:
1026+ assert (
1027+ type_args is None or len (type_args ) == 1
1028+ ), "OrderedSet should have exactly one type argument"
1029+ # Retrieve the type arguments from the class
1030+ type_arg = type_args [0 ] if type_args else None
1031+
1032+ if isinstance (value , CBORTag ) and value .tag == 258 :
1033+ if isclass (type_arg ) and issubclass (type_arg , CBORSerializable ):
1034+ value .value = [type_arg .from_primitive (v ) for v in value .value ]
1035+ return cls (value .value , use_tag = True )
1036+
1037+ if isinstance (value , (list , tuple , set )):
1038+ if isclass (type_arg ) and issubclass (type_arg , CBORSerializable ):
1039+ value = [type_arg .from_primitive (v ) for v in value ]
1040+ return cls (list (value ), use_tag = False )
1041+
1042+ raise ValueError (f"Cannot deserialize { value } { cls } )
1043+
1044+
1045+ class NonEmptyOrderedSet (OrderedSet [T ]):
1046+ def __init__ (self , iterable : Optional [List [T ]] = None , use_tag : bool = True ):
1047+ super ().__init__ (iterable , use_tag )
1048+
1049+ def validate (self ):
1050+ if not self :
1051+ raise ValueError ("NonEmptyOrderedSet cannot be empty" )
1052+
1053+ @classmethod
1054+ def from_primitive (
1055+ cls : Type [NonEmptyOrderedSet [T ]],
1056+ value : Primitive ,
1057+ type_args : Optional [tuple ] = None ,
1058+ ) -> NonEmptyOrderedSet [T ]:
1059+ result = cast (NonEmptyOrderedSet [T ], super ().from_primitive (value , type_args ))
1060+ if not result :
1061+ raise ValueError ("NonEmptyOrderedSet cannot be empty" )
1062+ return result
0 commit comments