31
31
A boolean; or
32
32
Null.
33
33
"""
34
- from typing import Any , Dict , List , Optional , Tuple , Union , cast
34
+ from typing import Any , Optional , Union , cast
35
35
36
36
from mypy_extensions import mypyc_attr
37
37
63
63
# MappingDataType = Dict[str, Union[PropType, List[PropsType]]]
64
64
# was: Union[str, MappingDataType, List[MappingDataType]]
65
65
JsonDataType = Any
66
- AtomicPropType = Union [None , str , int , float , bool , "Schema" , List [str ], List ["Field" ]]
66
+ AtomicPropType = Union [None , str , int , float , bool , "Schema" , list [str ], list ["Field" ]]
67
67
PropType = Union [
68
68
AtomicPropType ,
69
- Dict [str , AtomicPropType ],
70
- List [ Dict [str , AtomicPropType ]],
69
+ dict [str , AtomicPropType ],
70
+ list [ dict [str , AtomicPropType ]],
71
71
]
72
- PropsType = Dict [str , PropType ]
72
+ PropsType = dict [str , PropType ]
73
73
74
74
FIELD_RESERVED_PROPS = ("default" , "name" , "doc" , "order" , "type" )
75
75
@@ -189,7 +189,7 @@ class Names:
189
189
"""Track name set and default namespace during parsing."""
190
190
191
191
def __init__ (self , default_namespace : Optional [str ] = None ) -> None :
192
- self .names : Dict [str , NamedSchema ] = {}
192
+ self .names : dict [str , NamedSchema ] = {}
193
193
self .default_namespace = default_namespace
194
194
195
195
def has_name (self , name_attr : str , space_attr : Optional [str ]) -> bool :
@@ -281,7 +281,7 @@ def __init__(
281
281
default : Optional [Any ] = None ,
282
282
order : Optional [str ] = None ,
283
283
names : Optional [Names ] = None ,
284
- doc : Optional [Union [str , List [str ]]] = None ,
284
+ doc : Optional [Union [str , list [str ]]] = None ,
285
285
other_props : Optional [PropsType ] = None ,
286
286
) -> None :
287
287
# Ensure valid ctor args
@@ -361,9 +361,9 @@ def __init__(
361
361
self ,
362
362
name : str ,
363
363
namespace : Optional [str ],
364
- symbols : List [str ],
364
+ symbols : list [str ],
365
365
names : Optional [Names ] = None ,
366
- doc : Optional [Union [str , List [str ]]] = None ,
366
+ doc : Optional [Union [str , list [str ]]] = None ,
367
367
other_props : Optional [PropsType ] = None ,
368
368
) -> None :
369
369
# Ensure valid ctor args
@@ -384,8 +384,8 @@ def __init__(
384
384
385
385
# read-only properties
386
386
@property
387
- def symbols (self ) -> List [str ]:
388
- return cast (List [str ], self .get_prop ("symbols" ))
387
+ def symbols (self ) -> list [str ]:
388
+ return cast (list [str ], self .get_prop ("symbols" ))
389
389
390
390
391
391
#
@@ -473,7 +473,7 @@ def __init__(
473
473
names : Names ,
474
474
name : str ,
475
475
namespace : Optional [str ] = None ,
476
- doc : Optional [Union [str , List [str ]]] = None ,
476
+ doc : Optional [Union [str , list [str ]]] = None ,
477
477
other_props : Optional [PropsType ] = None ,
478
478
) -> None :
479
479
"""Create a NamedMapSchema object."""
@@ -505,8 +505,8 @@ def values(self) -> Schema:
505
505
return cast (Schema , self .get_prop ("values" ))
506
506
507
507
508
- def _build_schema_objects (schemas : List [JsonDataType ], names : Names ) -> List [Schema ]:
509
- schema_objects : List [Schema ] = []
508
+ def _build_schema_objects (schemas : list [JsonDataType ], names : Names ) -> list [Schema ]:
509
+ schema_objects : list [Schema ] = []
510
510
for schema in schemas :
511
511
if isinstance (schema , str ) and names .has_name (schema , None ):
512
512
new_schema = cast (Schema , names .get_name (schema , None ))
@@ -536,7 +536,7 @@ class UnionSchema(Schema):
536
536
537
537
def __init__ (
538
538
self ,
539
- schemas : List [JsonDataType ],
539
+ schemas : list [JsonDataType ],
540
540
names : Names ,
541
541
) -> None :
542
542
"""
@@ -558,7 +558,7 @@ def __init__(
558
558
559
559
# read-only properties
560
560
@property
561
- def schemas (self ) -> List [Schema ]:
561
+ def schemas (self ) -> list [Schema ]:
562
562
"""Avro schemas composing the Union type."""
563
563
return self ._schemas
564
564
@@ -568,11 +568,11 @@ class NamedUnionSchema(NamedSchema):
568
568
569
569
def __init__ (
570
570
self ,
571
- schemas : List [JsonDataType ],
571
+ schemas : list [JsonDataType ],
572
572
names : Names ,
573
573
name : str ,
574
574
namespace : Optional [str ] = None ,
575
- doc : Optional [Union [str , List [str ]]] = None ,
575
+ doc : Optional [Union [str , list [str ]]] = None ,
576
576
):
577
577
"""
578
578
Initialize a new NamedUnionSchema.
@@ -595,16 +595,16 @@ def __init__(
595
595
596
596
# read-only properties
597
597
@property
598
- def schemas (self ) -> List [Schema ]:
598
+ def schemas (self ) -> list [Schema ]:
599
599
return self ._schemas
600
600
601
601
602
602
class RecordSchema (NamedSchema ):
603
603
@staticmethod
604
- def make_field_objects (field_data : List [PropsType ], names : Names ) -> List [Field ]:
604
+ def make_field_objects (field_data : list [PropsType ], names : Names ) -> list [Field ]:
605
605
"""We're going to need to make message parameters too."""
606
- field_objects : List [Field ] = []
607
- parsed_fields : Dict [str , PropsType ] = {}
606
+ field_objects : list [Field ] = []
607
+ parsed_fields : dict [str , PropsType ] = {}
608
608
for field in field_data :
609
609
if hasattr (field , "get" ) and callable (field .get ):
610
610
atype = field .get ("type" )
@@ -622,7 +622,7 @@ def make_field_objects(field_data: List[PropsType], names: Names) -> List[Field]
622
622
doc = field .get ("doc" )
623
623
if not (doc is None or isinstance (doc , (list , str ))):
624
624
raise SchemaParseException ('"doc" must be a string, list of strings, or None' )
625
- doc = cast (Union [str , List [str ], None ], doc )
625
+ doc = cast (Union [str , list [str ], None ], doc )
626
626
other_props = get_other_props (field , FIELD_RESERVED_PROPS )
627
627
new_field = Field (atype , name , has_default , default , order , names , doc , other_props )
628
628
parsed_fields [new_field .name ] = field
@@ -635,10 +635,10 @@ def __init__(
635
635
self ,
636
636
name : str ,
637
637
namespace : Optional [str ],
638
- fields : List [PropsType ],
638
+ fields : list [PropsType ],
639
639
names : Names ,
640
640
schema_type : str = "record" ,
641
- doc : Optional [Union [str , List [str ]]] = None ,
641
+ doc : Optional [Union [str , list [str ]]] = None ,
642
642
other_props : Optional [PropsType ] = None ,
643
643
) -> None :
644
644
# Ensure valid ctor args
@@ -663,14 +663,14 @@ def __init__(
663
663
664
664
# read-only properties
665
665
@property
666
- def fields (self ) -> List [Field ]:
667
- return cast (List [Field ], self .get_prop ("fields" ))
666
+ def fields (self ) -> list [Field ]:
667
+ return cast (list [Field ], self .get_prop ("fields" ))
668
668
669
669
670
670
#
671
671
# Module Methods
672
672
#
673
- def get_other_props (all_props : PropsType , reserved_props : Tuple [str , ...]) -> Optional [PropsType ]:
673
+ def get_other_props (all_props : PropsType , reserved_props : tuple [str , ...]) -> Optional [PropsType ]:
674
674
"""
675
675
Retrieve the non-reserved properties from a dictionary of properties.
676
676
@@ -690,7 +690,7 @@ def make_avsc_object(json_data: JsonDataType, names: Optional[Names] = None) ->
690
690
if names is None :
691
691
names = Names ()
692
692
693
- if isinstance (json_data , Dict ) and json_data .get ("name" ) == "org.w3id.cwl.salad.Any" :
693
+ if isinstance (json_data , dict ) and json_data .get ("name" ) == "org.w3id.cwl.salad.Any" :
694
694
del names .names ["org.w3id.cwl.salad.Any" ]
695
695
elif not names .has_name ("org.w3id.cwl.salad.Any" , None ):
696
696
EnumSchema ("org.w3id.cwl.salad.Any" , None , ["Any" ], names = names )
@@ -723,15 +723,15 @@ def make_avsc_object(json_data: JsonDataType, names: Optional[Names] = None) ->
723
723
raise SchemaParseException (
724
724
f'"symbols" for type enum must be a list of strings: { json_data } '
725
725
)
726
- symbols = cast (List [str ], symbols )
726
+ symbols = cast (list [str ], symbols )
727
727
return EnumSchema (name , namespace , symbols , names , doc , other_props )
728
728
if atype in ["record" , "error" ]:
729
729
fields = json_data .get ("fields" , [])
730
730
if not isinstance (fields , list ):
731
731
raise SchemaParseException (
732
732
f'"fields" for type { atype } must be a list of mappings: { json_data } '
733
733
)
734
- fields = cast (List [PropsType ], fields )
734
+ fields = cast (list [PropsType ], fields )
735
735
return RecordSchema (name , namespace , fields , names , atype , doc , other_props )
736
736
raise SchemaParseException (f"Unknown Named Type: { atype } " )
737
737
if atype in VALID_TYPES :
@@ -774,7 +774,7 @@ def make_avsc_object(json_data: JsonDataType, names: Optional[Names] = None) ->
774
774
raise SchemaParseException (fail_msg )
775
775
776
776
777
- def is_subtype (types : Dict [str , Any ], existing : PropType , new : PropType ) -> bool :
777
+ def is_subtype (types : dict [str , Any ], existing : PropType , new : PropType ) -> bool :
778
778
"""Check if a new type specification is compatible with an existing type spec."""
779
779
if existing == new :
780
780
return True
@@ -804,9 +804,9 @@ def is_subtype(types: Dict[str, Any], existing: PropType, new: PropType) -> bool
804
804
if existing .get ("type" ) == "enum" and new .get ("type" ) == "enum" :
805
805
return is_subtype (types , existing ["symbols" ], new ["symbols" ])
806
806
if existing .get ("type" ) == "record" and new .get ("type" ) == "record" :
807
- for new_field in cast (List [ Dict [str , Any ]], new ["fields" ]):
807
+ for new_field in cast (list [ dict [str , Any ]], new ["fields" ]):
808
808
new_field_missing = True
809
- for existing_field in cast (List [ Dict [str , Any ]], existing ["fields" ]):
809
+ for existing_field in cast (list [ dict [str , Any ]], existing ["fields" ]):
810
810
if new_field ["name" ] == existing_field ["name" ]:
811
811
if not is_subtype (types , existing_field ["type" ], new_field ["type" ]):
812
812
return False
0 commit comments