| 918 | |
| 919 | |
| 920 | class Dict(NestedContainer, Mapping): |
| 921 | klass = dict |
| 922 | |
| 923 | def __init__(self, /, *args: Any, **kwargs: Any): |
| 924 | if args: |
| 925 | assert not kwargs |
| 926 | if len(args) == 1: |
| 927 | args = args[0] |
| 928 | if isinstance(args, dict): # type: ignore |
| 929 | args = tuple(itertools.chain(*args.items())) # type: ignore |
| 930 | elif isinstance(args, (list, tuple)): |
| 931 | if all( |
| 932 | len(el) == 2 if isinstance(el, (list, tuple)) else False |
| 933 | for el in args |
| 934 | ): |
| 935 | args = tuple(itertools.chain(*args)) |
| 936 | else: |
| 937 | raise ValueError("Invalid argument provided") |
| 938 | |
| 939 | if len(args) % 2 != 0: |
| 940 | raise ValueError("Invalid number of arguments provided") |
| 941 | |
| 942 | elif kwargs: |
| 943 | assert not args |
| 944 | args = tuple(itertools.chain(*kwargs.items())) |
| 945 | |
| 946 | super().__init__(*args) |
| 947 | |
| 948 | def __repr__(self): |
| 949 | values = ", ".join(f"{k}: {v}" for k, v in batched(self.args, 2, strict=True)) |
| 950 | return f"Dict({values})" |
| 951 | |
| 952 | def substitute( |
| 953 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 954 | ) -> Dict: |
| 955 | subs_filtered = { |
| 956 | k: v for k, v in subs.items() if k in self.dependencies and k != v |
| 957 | } |
| 958 | if not subs_filtered: |
| 959 | return self |
| 960 | |
| 961 | new_args = [] |
| 962 | for arg in self.args: |
| 963 | new_arg = ( |
| 964 | arg.substitute(subs_filtered) |
| 965 | if isinstance(arg, (GraphNode, TaskRef)) |
| 966 | else arg |
| 967 | ) |
| 968 | new_args.append(new_arg) |
| 969 | return type(self)(new_args) |
| 970 | |
| 971 | def __iter__(self): |
| 972 | yield from self.args[::2] |
| 973 | |
| 974 | def __len__(self): |
| 975 | return len(self.args) // 2 |
| 976 | |
| 977 | def __getitem__(self, key): |
no outgoing calls