Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/experimental.py: 11%

1""" 

2Experimental CLI commands. 

3 

4These commands provide access to experimental features that are subject to change. 

5""" 

6 

7from __future__ import annotations 1 ctx1a

8 

9import importlib.metadata as md 1 ctx1a

10 

11from prefect._experimental.plugins.manager import ENTRYPOINTS_GROUP 1 ctx1a

12from prefect.cli._types import PrefectTyper 1 ctx1a

13from prefect.cli.root import app 1 ctx1a

14from prefect.settings import get_current_settings 1 ctx1a

15 

16experimental_app: PrefectTyper = PrefectTyper( 1 ctx1a

17 name="experimental", 

18 help="Access experimental features (subject to change).", 

19 hidden=True, # Hidden from main help, but accessible 

20) 

21app.add_typer(experimental_app) 1 ctx1a

22 

23plugins_app: PrefectTyper = PrefectTyper( 1 ctx1a

24 name="plugins", 

25 help="Plugin system diagnostics.", 

26) 

27experimental_app.add_typer(plugins_app) 1 ctx1a

28 

29 

30@plugins_app.command("diagnose") 1 ctx1a

31async def diagnose(): 1 ctx1a

32 """ 

33 Diagnose the experimental plugin system. 

34 

35 This command provides information about: 

36 - Whether the plugin system is enabled 

37 - What plugins are discoverable via entry points 

38 - Current configuration (timeouts, allow/deny lists) 

39 

40 Note: This runs the plugin discovery but does not execute hooks. 

41 Use safe mode (PREFECT_EXPERIMENTS_PLUGINS_SAFE_MODE=1) to test plugin loading 

42 without executing hooks. 

43 """ 

44 from prefect._experimental.plugins import run_startup_hooks 

45 

46 app.console.print("\n[bold]Prefect Experimental Plugin System Diagnostics[/bold]\n") 

47 

48 # Check if enabled 

49 settings = get_current_settings().experiments.plugins 

50 enabled = settings.enabled 

51 app.console.print(f"Enabled: [{'green' if enabled else 'red'}]{enabled}[/]") 

52 

53 if not enabled: 

54 app.console.print("\n[yellow]Plugin system is disabled.[/yellow]") 

55 app.console.print( 

56 "Set [cyan]PREFECT_EXPERIMENTS_PLUGINS_ENABLED=1[/cyan] to enable.\n" 

57 ) 

58 return 

59 

60 # Show configuration 

61 timeout = settings.setup_timeout_seconds 

62 allow = settings.allow 

63 deny = settings.deny 

64 strict = settings.strict 

65 safe = settings.safe_mode 

66 

67 app.console.print(f"Timeout: {timeout}s") 

68 app.console.print(f"Strict mode: {strict}") 

69 app.console.print(f"Safe mode: {safe}") 

70 app.console.print(f"Allow list: {allow or 'None'}") 

71 app.console.print(f"Deny list: {deny or 'None'}") 

72 

73 # Discover entry points 

74 app.console.print( 

75 f"\n[bold]Discoverable Plugins (entry point group: {ENTRYPOINTS_GROUP})[/bold]\n" 

76 ) 

77 

78 entry_points = list(md.entry_points(group=ENTRYPOINTS_GROUP)) 

79 

80 if not entry_points: 

81 app.console.print("[yellow]No plugins found.[/yellow]\n") 

82 return 

83 

84 for ep in entry_points: 

85 filtered = False 

86 reason = None 

87 if allow and ep.name not in allow: 

88 filtered = True 

89 reason = "not in allow list" 

90 elif deny and ep.name in deny: 

91 filtered = True 

92 reason = "in deny list" 

93 

94 status_color = "red" if filtered else "green" 

95 status = f"[{status_color}]{'filtered' if filtered else 'active'}[/]" 

96 

97 app.console.print(f" • {ep.name}: {status}") 

98 app.console.print(f" Module: {ep.value}") 

99 

100 if filtered and reason: 

101 app.console.print(f" Reason: {reason}") 

102 

103 # Try to load and get version requirement 

104 try: 

105 plugin = ep.load() 

106 requires = getattr(plugin, "PREFECT_PLUGIN_API_REQUIRES", ">=0.1,<1") 

107 app.console.print(f" API requirement: {requires}") 

108 except Exception as e: 

109 app.console.print(f" [red]Failed to load: {e}[/]") 

110 

111 app.console.print() 

112 

113 # Run startup hooks to show what they do 

114 if not safe: 

115 app.console.print("[bold]Running Startup Hooks[/bold]\n") 

116 

117 from prefect import __version__ 

118 from prefect._experimental.plugins.spec import HookContext 

119 from prefect.logging import get_logger 

120 

121 ctx = HookContext( 

122 prefect_version=__version__, 

123 api_url=get_current_settings().api.url, 

124 logger_factory=get_logger, 

125 ) 

126 summaries = await run_startup_hooks(ctx) 

127 

128 if summaries: 

129 for summary in summaries: 

130 status = "[red]error[/]" if summary.error else "[green]success[/]" 

131 app.console.print(f" • {summary.plugin}: {status}") 

132 

133 if summary.error: 

134 app.console.print(f" Error: {summary.error}") 

135 elif summary.env_preview: 

136 app.console.print( 

137 f" Environment variables: {len(summary.env_preview)}" 

138 ) 

139 for k, v in summary.env_preview.items(): 

140 app.console.print(f" {k}={v}") 

141 

142 if summary.note: 

143 app.console.print(f" Note: {summary.note}") 

144 else: 

145 app.console.print(" No changes") 

146 

147 app.console.print() 

148 else: 

149 app.console.print("[yellow]No plugins executed.[/yellow]\n") 

150 else: 

151 app.console.print( 

152 "\n[yellow]Safe mode enabled - skipping hook execution.[/yellow]" 

153 ) 

154 app.console.print( 

155 "Set [cyan]PREFECT_EXPERIMENTS_PLUGINS_SAFE_MODE=0[/cyan] to execute hooks.\n" 

156 )