Coverage for /usr/local/lib/python3.12/site-packages/prefect/plugins.py: 78%
27 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 11:21 +0000
1"""
2Utilities for loading plugins that extend Prefect's functionality.
4Plugins are detected by entry point definitions in package setup files.
6Currently supported entrypoints:
7 - prefect.collections: Identifies this package as a Prefect collection that
8 should be imported when Prefect is imported.
9"""
11from importlib.metadata import EntryPoints, entry_points 1b
12from types import ModuleType 1b
13from typing import Any, Union 1b
15import prefect.settings 1b
17_collections: Union[None, dict[str, Union[ModuleType, Exception]]] = None 1b
20def safe_load_entrypoints(entrypoints: EntryPoints) -> dict[str, Union[Exception, Any]]: 1b
21 """
22 Load entry points for a group capturing any exceptions that occur.
23 """
24 # TODO: `load()` claims to return module types but could return arbitrary types
25 # too. We can cast the return type if we want to be more correct. We may
26 # also want to validate the type for the group for entrypoints that have
27 # a specific type we expect.
29 results: dict[str, Union[Exception, Any]] = {} 1a
31 for entrypoint in entrypoints: 1a
32 result = None 1a
33 try: 1a
34 result = entrypoint.load() 1a
35 except Exception as exc:
36 result = exc
38 results[entrypoint.name or entrypoint.value] = result 1a
40 return results 1a
43def load_prefect_collections() -> dict[str, Union[ModuleType, Exception]]: 1b
44 """
45 Load all Prefect collections that define an entrypoint in the group
46 `prefect.collections`.
47 """
48 global _collections
50 if _collections is not None: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true1a
51 return _collections
53 collection_entrypoints: EntryPoints = entry_points(group="prefect.collections") 1a
54 collections: dict[str, Union[Exception, Any]] = safe_load_entrypoints( 1a
55 collection_entrypoints
56 )
58 # TODO: Consider the utility of this once we've established this pattern.
59 # We cannot use a logger here because logging is not yet initialized.
60 # It would be nice if logging was initialized so we could log failures
61 # at least.
62 for name, result in collections.items(): 1a
63 if isinstance(result, Exception): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true1a
64 print(
65 # TODO: Use exc_info if we have a logger
66 f"Warning! Failed to load collection {name!r}:"
67 f" {type(result).__name__}: {result}"
68 )
69 else:
70 if prefect.settings.PREFECT_DEBUG_MODE: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true1a
71 print(f"Loaded collection {name!r}.")
73 _collections = collections 1a
74 return collections 1a