88import os .path
99import stat
1010import sys
11- from collections .abc import MutableSequence , Sequence
11+ from collections .abc import MutableMapping , MutableSequence , Sequence
1212from pathlib import Path
13- from typing import Any , Callable , Dict , List , MutableMapping , Optional , Set , Union
13+ from typing import Any , Callable , Optional , Union
1414
1515import ruamel .yaml
1616from ruamel .yaml .comments import CommentedMap # for consistent sort order
2727yaml .default_flow_style = False
2828
2929
30- def parse_args (args : List [str ]) -> argparse .Namespace :
30+ def parse_args (args : list [str ]) -> argparse .Namespace :
3131 """Argument parser."""
3232 parser = argparse .ArgumentParser (
3333 description = "Tool to upgrade CWL documents from one version to another. "
@@ -59,7 +59,7 @@ def parse_args(args: List[str]) -> argparse.Namespace:
5959 return parser .parse_args (args )
6060
6161
62- def main (args : Optional [List [str ]] = None ) -> int :
62+ def main (args : Optional [list [str ]] = None ) -> int :
6363 """Hook to set the args."""
6464 if not args :
6565 args = sys .argv [1 :]
@@ -68,7 +68,7 @@ def main(args: Optional[List[str]] = None) -> int:
6868
6969def run (args : argparse .Namespace ) -> int :
7070 """Main function."""
71- imports : Set [str ] = set ()
71+ imports : set [str ] = set ()
7272 if args .dir and not os .path .exists (args .dir ):
7373 os .makedirs (args .dir )
7474 for path in args .inputs :
@@ -107,7 +107,7 @@ def upgrade_document(
107107 document : Any ,
108108 output_dir : str ,
109109 target_version : Optional [str ] = "latest" ,
110- imports : Optional [Set [str ]] = None ,
110+ imports : Optional [set [str ]] = None ,
111111) -> Any :
112112 if imports is None :
113113 imports = set ()
@@ -210,7 +210,7 @@ def write_cwl_document(document: Any, name: str, dirname: str) -> None:
210210
211211
212212def process_imports (
213- document : Any , imports : Set [str ], updater : Callable [[Any , str ], Any ], outdir : str
213+ document : Any , imports : set [str ], updater : Callable [[Any , str ], Any ], outdir : str
214214) -> None :
215215 """Find any '$import's and process them."""
216216 if isinstance (document , CommentedMap ):
@@ -481,10 +481,10 @@ def _v1_1_to_v1_2(document: CommentedMap, outdir: str) -> CommentedMap:
481481 return document
482482
483483
484- def cleanup_v1_0_input_bindings (document : Dict [str , Any ]) -> None :
484+ def cleanup_v1_0_input_bindings (document : dict [str , Any ]) -> None :
485485 """In v1.1 Workflow or ExpressionTool level inputBindings are deprecated."""
486486
487- def cleanup (inp : Dict [str , Any ]) -> None :
487+ def cleanup (inp : dict [str , Any ]) -> None :
488488 """Serialize non loadContents fields and add that to the doc."""
489489 if "inputBinding" in inp :
490490 bindings = inp ["inputBinding" ]
@@ -505,10 +505,10 @@ def cleanup(inp: Dict[str, Any]) -> None:
505505 cleanup (inputs [input_name ])
506506
507507
508- def move_up_loadcontents (document : Dict [str , Any ]) -> None :
508+ def move_up_loadcontents (document : dict [str , Any ]) -> None :
509509 """Promote 'loadContents' up a level for CWL v1.1."""
510510
511- def cleanup (inp : Dict [str , Any ]) -> None :
511+ def cleanup (inp : dict [str , Any ]) -> None :
512512 """Move loadContents to the preferred location."""
513513 if "inputBinding" in inp :
514514 bindings = inp ["inputBinding" ]
@@ -525,7 +525,7 @@ def cleanup(inp: Dict[str, Any]) -> None:
525525 cleanup (inputs [input_name ])
526526
527527
528- def upgrade_v1_0_hints_and_reqs (document : Dict [str , Any ]) -> None :
528+ def upgrade_v1_0_hints_and_reqs (document : dict [str , Any ]) -> None :
529529 """Rename some pre-v1.1 extensions to their official CWL v1.1 names."""
530530 for extra in ("requirements" , "hints" ):
531531 if extra in document :
@@ -555,7 +555,7 @@ def upgrade_v1_0_hints_and_reqs(document: Dict[str, Any]) -> None:
555555 )
556556
557557
558- def has_hint_or_req (document : Dict [str , Any ], name : str ) -> bool :
558+ def has_hint_or_req (document : dict [str , Any ], name : str ) -> bool :
559559 """Detects an existing named hint or requirement."""
560560 for extra in ("requirements" , "hints" ):
561561 if extra in document :
@@ -571,7 +571,7 @@ def has_hint_or_req(document: Dict[str, Any], name: str) -> bool:
571571 return False
572572
573573
574- def workflow_clean (document : Dict [str , Any ]) -> None :
574+ def workflow_clean (document : dict [str , Any ]) -> None :
575575 """Transform draft-3 style Workflows to more idiomatic v1.0"""
576576 input_output_clean (document )
577577 hints_and_requirements_clean (document )
@@ -652,7 +652,7 @@ def workflow_clean(document: Dict[str, Any]) -> None:
652652 document ["steps" ] = new_steps
653653
654654
655- def input_output_clean (document : Dict [str , Any ]) -> None :
655+ def input_output_clean (document : dict [str , Any ]) -> None :
656656 """Transform draft-3 style input/output listings into idiomatic v1.0."""
657657 for param_type in ["inputs" , "outputs" ]:
658658 if param_type not in document :
@@ -701,7 +701,7 @@ def array_type_raise_sf(param: MutableMapping[str, Any]) -> None:
701701 del typ ["secondaryFiles" ]
702702
703703
704- def hints_and_requirements_clean (document : Dict [str , Any ]) -> None :
704+ def hints_and_requirements_clean (document : dict [str , Any ]) -> None :
705705 """Transform draft-3 style hints/reqs into idiomatic v1.0 hints/reqs."""
706706 for section in ["hints" , "requirements" ]:
707707 if section in document :
@@ -736,13 +736,13 @@ def hints_and_requirements_clean(document: Dict[str, Any]) -> None:
736736 document [section ] = new_section
737737
738738
739- def shorten_type (type_obj : Union [str , List [Any ]]) -> Union [str , List [Any ]]:
739+ def shorten_type (type_obj : Union [str , list [Any ]]) -> Union [str , list [Any ]]:
740740 """Transform draft-3 style type declarations into idiomatic v1.0 types."""
741741 if isinstance (type_obj , str ) or not isinstance (type_obj , Sequence ):
742742 return type_obj
743743 new_type = [] # type: List[str]
744744 for entry in type_obj : # find arrays that we can shorten and do so
745- if isinstance (entry , Dict ):
745+ if isinstance (entry , dict ):
746746 if entry ["type" ] == "array" and isinstance (entry ["items" ], str ):
747747 entry = entry ["items" ] + "[]"
748748 elif entry ["type" ] == "enum" :
@@ -759,7 +759,7 @@ def shorten_type(type_obj: Union[str, List[Any]]) -> Union[str, List[Any]]:
759759 return new_type
760760
761761
762- def clean_secondary_files (document : Dict [str , Any ]) -> None :
762+ def clean_secondary_files (document : dict [str , Any ]) -> None :
763763 """Cleanup for secondaryFiles"""
764764 if "secondaryFiles" in document :
765765 for i , sfile in enumerate (document ["secondaryFiles" ]):
@@ -769,7 +769,7 @@ def clean_secondary_files(document: Dict[str, Any]) -> None:
769769 ).replace (".path" , ".location" )
770770
771771
772- def sort_v1_0 (document : Dict [str , Any ]) -> CommentedMap :
772+ def sort_v1_0 (document : dict [str , Any ]) -> CommentedMap :
773773 """Sort the sections of the CWL document in a more meaningful order."""
774774 keyorder = [
775775 "cwlVersion" ,
@@ -800,7 +800,7 @@ def sort_v1_0(document: Dict[str, Any]) -> CommentedMap:
800800 )
801801
802802
803- def sort_enum (enum : Dict [str , Any ]) -> Dict [str , Any ]:
803+ def sort_enum (enum : dict [str , Any ]) -> dict [str , Any ]:
804804 """Sort the enum type definitions in a more meaningful order."""
805805 keyorder = ["type" , "name" , "label" , "symbols" , "inputBinding" ]
806806 return CommentedMap (
@@ -811,7 +811,7 @@ def sort_enum(enum: Dict[str, Any]) -> Dict[str, Any]:
811811 )
812812
813813
814- def sort_input_or_output (io_def : Dict [str , Any ]) -> Dict [str , Any ]:
814+ def sort_input_or_output (io_def : dict [str , Any ]) -> dict [str , Any ]:
815815 """Sort the input definitions in a more meaningful order."""
816816 keyorder = [
817817 "label" ,
0 commit comments