Load entrypoints and instantiate chunkmanagers only once.
(
entrypoints: Sequence[EntryPoint],
)
| 70 | |
| 71 | |
| 72 | def load_chunkmanagers( |
| 73 | entrypoints: Sequence[EntryPoint], |
| 74 | ) -> dict[str, ChunkManagerEntrypoint[Any]]: |
| 75 | """Load entrypoints and instantiate chunkmanagers only once.""" |
| 76 | |
| 77 | loaded_entrypoints = {} |
| 78 | for entrypoint in entrypoints: |
| 79 | try: |
| 80 | loaded_entrypoints[entrypoint.name] = entrypoint.load() |
| 81 | except ModuleNotFoundError as e: |
| 82 | emit_user_level_warning( |
| 83 | f"Failed to load chunk manager entrypoint {entrypoint.name} due to {e}. Skipping.", |
| 84 | ) |
| 85 | |
| 86 | available_chunkmanagers = { |
| 87 | name: chunkmanager() |
| 88 | for name, chunkmanager in loaded_entrypoints.items() |
| 89 | if chunkmanager.available |
| 90 | } |
| 91 | return available_chunkmanagers |
| 92 | |
| 93 | |
| 94 | def guess_chunkmanager( |
searching dependent graphs…