Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/_types.py: 71%
68 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1"""
2Custom Prefect CLI types
3"""
5import asyncio 1a
6import functools 1a
7import sys 1a
8from datetime import datetime 1a
9from typing import TYPE_CHECKING, Any, Callable, List, Optional 1a
11import typer 1a
12from rich.console import Console 1a
13from rich.theme import Theme 1a
15from prefect._internal.compatibility.deprecated import generate_deprecation_message 1a
16from prefect.cli._utilities import with_cli_exception_handling 1a
17from prefect.settings import get_current_settings 1a
18from prefect.utilities.asyncutils import is_async_fn 1a
20if TYPE_CHECKING: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true1a
21 from prefect.settings.legacy import Setting
24def SettingsOption(setting: "Setting", *args: Any, **kwargs: Any) -> Any: 1a
25 """Custom `typer.Option` factory to load the default value from settings"""
27 return typer.Option( 1a
28 # The default is dynamically retrieved
29 setting.value,
30 *args,
31 # Typer shows "(dynamic)" by default. We'd like to actually show the value
32 # that would be used if the parameter is not specified and a reference if the
33 # source is from the environment or profile, but typer does not support this
34 # yet. See https://github.com/tiangolo/typer/issues/354
35 show_default=f"from {setting.name}",
36 **kwargs,
37 )
40def SettingsArgument(setting: "Setting", *args: Any, **kwargs: Any) -> Any: 1a
41 """Custom `typer.Argument` factory to load the default value from settings"""
43 # See comments in `SettingsOption`
44 return typer.Argument(
45 setting.value,
46 *args,
47 show_default=f"from {setting.name}",
48 **kwargs,
49 )
52def with_deprecated_message( 1a
53 warning: str,
54) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
55 def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
56 @functools.wraps(fn)
57 def wrapper(*args: Any, **kwargs: Any) -> Any:
58 print("WARNING:", warning, file=sys.stderr, flush=True)
59 return fn(*args, **kwargs)
61 return wrapper
63 return decorator
66class PrefectTyper(typer.Typer): 1a
67 """
68 Wraps commands created by `Typer` to support async functions and handle errors.
69 """
71 console: Console 1a
73 def __init__( 1a
74 self,
75 *args: Any,
76 deprecated: bool = False,
77 deprecated_start_date: Optional[datetime] = None,
78 deprecated_help: str = "",
79 deprecated_name: str = "",
80 **kwargs: Any,
81 ):
82 super().__init__(*args, **kwargs) 1a
84 self.deprecated = deprecated 1a
85 if self.deprecated: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true1a
86 if not deprecated_name:
87 raise ValueError("Provide the name of the deprecated command group.")
88 self.deprecated_message: str = generate_deprecation_message(
89 name=f"The {deprecated_name!r} command group",
90 start_date=deprecated_start_date,
91 help=deprecated_help,
92 )
94 self.console = Console( 1a
95 highlight=False,
96 theme=Theme({"prompt.choices": "bold blue"}),
97 color_system="auto" if get_current_settings().cli.colors else None,
98 )
100 def add_typer( 1a
101 self,
102 typer_instance: "PrefectTyper",
103 *args: Any,
104 no_args_is_help: bool = True,
105 aliases: Optional[list[str]] = None,
106 **kwargs: Any,
107 ) -> None:
108 """
109 This will cause help to be default command for all sub apps unless specifically stated otherwise, opposite of before.
110 """
111 if aliases: 1a
112 for alias in aliases: 1a
113 super().add_typer( 1a
114 typer_instance,
115 *args,
116 name=alias,
117 no_args_is_help=no_args_is_help,
118 hidden=True,
119 **kwargs,
120 )
122 return super().add_typer( 1a
123 typer_instance, *args, no_args_is_help=no_args_is_help, **kwargs
124 )
126 def command( 1a
127 self,
128 name: Optional[str] = None,
129 *args: Any,
130 aliases: Optional[List[str]] = None,
131 deprecated: bool = False,
132 deprecated_start_date: Optional[datetime] = None,
133 deprecated_help: str = "",
134 deprecated_name: str = "",
135 **kwargs: Any,
136 ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
137 """
138 Create a new command. If aliases are provided, the same command function
139 will be registered with multiple names.
141 Provide `deprecated=True` to mark the command as deprecated. If `deprecated=True`,
142 `deprecated_name` and `deprecated_start_date` must be provided.
143 """
145 def wrapper(original_fn: Callable[..., Any]) -> Callable[..., Any]: 1a
146 # click doesn't support async functions, so we wrap them in
147 # asyncio.run(). This has the advantage of keeping the function in
148 # the main thread, which means signal handling works for e.g. the
149 # server and workers. However, it means that async CLI commands can
150 # not directly call other async CLI commands (because asyncio.run()
151 # can not be called nested). In that (rare) circumstance, refactor
152 # the CLI command so its business logic can be invoked separately
153 # from its entrypoint.
154 if is_async_fn(original_fn): 1a
155 async_fn = original_fn 1a
157 @functools.wraps(original_fn) 1a
158 def sync_fn(*args: Any, **kwargs: Any) -> Any: 1a
159 return asyncio.run(async_fn(*args, **kwargs))
161 setattr(sync_fn, "aio", async_fn) 1a
162 wrapped_fn = sync_fn 1a
163 else:
164 wrapped_fn = original_fn 1a
166 wrapped_fn = with_cli_exception_handling(wrapped_fn) 1a
167 if deprecated: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true1a
168 if not deprecated_name or not deprecated_start_date:
169 raise ValueError(
170 "Provide the name of the deprecated command and a deprecation start date."
171 )
172 command_deprecated_message = generate_deprecation_message(
173 name=f"The {deprecated_name!r} command",
174 start_date=deprecated_start_date,
175 help=deprecated_help,
176 )
177 wrapped_fn = with_deprecated_message(command_deprecated_message)(
178 wrapped_fn
179 )
180 elif self.deprecated: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true1a
181 wrapped_fn = with_deprecated_message(self.deprecated_message)(
182 wrapped_fn
183 )
185 # register fn with its original name
186 command_decorator = super(PrefectTyper, self).command( 1a
187 name=name, *args, **kwargs
188 )
189 original_command = command_decorator(wrapped_fn) 1a
191 # register fn for each alias, e.g. @marvin_app.command(aliases=["r"])
192 if aliases: 1a
193 for alias in aliases: 1a
194 super(PrefectTyper, self).command( 1a
195 name=alias,
196 *args,
197 **{k: v for k, v in kwargs.items() if k != "aliases"},
198 )(wrapped_fn)
200 return original_command 1a
202 return wrapper 1a
204 def setup_console(self, soft_wrap: bool, prompt: bool) -> None: 1a
205 self.console = Console( 1a
206 highlight=False,
207 color_system="auto" if get_current_settings().cli.colors else None,
208 theme=Theme({"prompt.choices": "bold blue"}),
209 soft_wrap=not soft_wrap,
210 force_interactive=prompt,
211 )