(
dsk: dict[KeyType, GraphNode], keys: Iterable[KeyType]
)
| 1168 | |
| 1169 | |
| 1170 | def cull( |
| 1171 | dsk: dict[KeyType, GraphNode], keys: Iterable[KeyType] |
| 1172 | ) -> dict[KeyType, GraphNode]: |
| 1173 | if not isinstance(keys, (list, set, tuple)): |
| 1174 | raise TypeError( |
| 1175 | f"Expected list, set or tuple for keys, got {type(keys).__name__}" |
| 1176 | ) |
| 1177 | if len(keys) == len(dsk): |
| 1178 | return dsk |
| 1179 | work = set(keys) |
| 1180 | seen: set[KeyType] = set() |
| 1181 | dsk2 = {} |
| 1182 | wpop = work.pop |
| 1183 | wupdate = work.update |
| 1184 | sadd = seen.add |
| 1185 | while work: |
| 1186 | k = wpop() |
| 1187 | if k in seen or k not in dsk: |
| 1188 | continue |
| 1189 | sadd(k) |
| 1190 | dsk2[k] = v = dsk[k] |
| 1191 | wupdate(v.dependencies) |
| 1192 | return dsk2 |
| 1193 | |
| 1194 | |
| 1195 | @functools.cache |
no test coverage detected