Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/compatibility/deprecated.py: 28%
114 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 11:21 +0000
1"""
2Utilities for deprecated items.
4When a deprecated item is used, a warning will be displayed. Warnings may not be
5disabled with Prefect settings. Instead, the standard Python warnings filters can be
6used.
8Deprecated items require a start or end date. If a start date is given, the end date
9will be calculated 6 months later. Start and end dates are always in the format MMM YYYY
10e.g. Jan 2023.
11"""
13from __future__ import annotations 1a
15import datetime 1a
16import functools 1a
17import sys 1a
18import warnings 1a
19from typing import TYPE_CHECKING, Any, Callable, Optional, Union 1a
21from pydantic import BaseModel 1a
22from typing_extensions import ParamSpec, TypeAlias, TypeVar 1a
24from prefect.types._datetime import now 1a
25from prefect.utilities.callables import get_call_parameters 1a
26from prefect.utilities.importtools import ( 1a
27 AliasedModuleDefinition,
28 AliasedModuleFinder,
29 to_qualified_name,
30)
32P = ParamSpec("P") 1a
33R = TypeVar("R", infer_variance=True) 1a
34M = TypeVar("M", bound=BaseModel) 1a
35T = TypeVar("T") 1a
37# Note: A datetime is strongly preferred over a string, but a string is acceptable for
38# backwards compatibility until support is dropped from dateparser in Python 3.15.
39_AcceptableDate: TypeAlias = Optional[Union[datetime.datetime, str]] 1a
41DEPRECATED_WARNING = ( 1a
42 "{name} has been deprecated{when}. It will not be available in new releases after {end_date}."
43 " {help}"
44)
45DEPRECATED_MOVED_WARNING = ( 1a
46 "{name} has moved to {new_location}. It will not be available at the old import "
47 "path after {end_date}. {help}"
48)
49HUMAN_DATEFMT = "%b %Y" # e.g. Feb 2023 1a
50DEPRECATED_MODULE_ALIASES: list[AliasedModuleDefinition] = [] 1a
53class PrefectDeprecationWarning(DeprecationWarning): 1a
54 """
55 A deprecation warning.
56 """
59def _coerce_datetime( 1a
60 dt: Optional[_AcceptableDate],
61) -> Optional[datetime.datetime]:
62 if dt is None or isinstance(dt, datetime.datetime):
63 return dt
64 with warnings.catch_warnings():
65 warnings.filterwarnings("ignore", category=DeprecationWarning)
66 import dateparser
68 return dateparser.parse(dt)
71def generate_deprecation_message( 1a
72 name: str,
73 start_date: Optional[_AcceptableDate] = None,
74 end_date: Optional[_AcceptableDate] = None,
75 help: str = "",
76 when: str = "",
77) -> str:
78 start_date = _coerce_datetime(start_date)
79 end_date = _coerce_datetime(end_date)
81 if start_date is None and end_date is None:
82 raise ValueError(
83 "A start date is required if an end date is not provided. Suggested start"
84 f" date is {now('UTC').strftime(HUMAN_DATEFMT)}"
85 )
87 if end_date is None:
88 if TYPE_CHECKING:
89 assert start_date is not None
91 if sys.version_info >= (3, 13):
92 from whenever import PlainDateTime
94 end_date = (
95 PlainDateTime.from_py_datetime(start_date).add(months=6).py_datetime()
96 )
97 else:
98 import pendulum
100 end_date = pendulum.instance(start_date).add(months=6)
102 if when:
103 when = " when " + when
105 message = DEPRECATED_WARNING.format(
106 name=name, when=when, end_date=end_date.strftime(HUMAN_DATEFMT), help=help
107 )
108 return message.rstrip()
111def deprecated_callable( 1a
112 *,
113 start_date: Optional[_AcceptableDate] = None,
114 end_date: Optional[_AcceptableDate] = None,
115 stacklevel: int = 2,
116 help: str = "",
117) -> Callable[[Callable[P, R]], Callable[P, R]]:
118 def decorator(fn: Callable[P, R]) -> Callable[P, R]: 1a
119 @functools.wraps(fn) 1a
120 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 1a
121 message = generate_deprecation_message(
122 name=to_qualified_name(fn),
123 start_date=start_date,
124 end_date=end_date,
125 help=help,
126 )
127 warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)
128 return fn(*args, **kwargs)
130 return wrapper 1a
132 return decorator 1a
135def deprecated_class( 1a
136 *,
137 start_date: Optional[_AcceptableDate] = None,
138 end_date: Optional[_AcceptableDate] = None,
139 stacklevel: int = 2,
140 help: str = "",
141) -> Callable[[type[T]], type[T]]:
142 def decorator(cls: type[T]) -> type[T]:
143 original_init = cls.__init__
145 @functools.wraps(original_init)
146 def new_init(self: T, *args: Any, **kwargs: Any) -> None:
147 message = generate_deprecation_message(
148 name=to_qualified_name(cls),
149 start_date=start_date,
150 end_date=end_date,
151 help=help,
152 )
153 warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)
154 original_init(self, *args, **kwargs)
156 cls.__init__ = new_init
157 return cls
159 return decorator
162def deprecated_parameter( 1a
163 name: str,
164 *,
165 start_date: Optional[_AcceptableDate] = None,
166 end_date: Optional[_AcceptableDate] = None,
167 stacklevel: int = 2,
168 help: str = "",
169 when: Optional[Callable[[Any], bool]] = None,
170 when_message: str = "",
171) -> Callable[[Callable[P, R]], Callable[P, R]]:
172 """
173 Mark a parameter in a callable as deprecated.
175 Example:
177 ```python
179 @deprecated_parameter("y", when=lambda y: y is not None)
180 def foo(x, y = None):
181 return x + 1 + (y or 0)
182 ```
183 """
185 when = when or (lambda _: True)
187 def decorator(fn: Callable[P, R]) -> Callable[P, R]:
188 @functools.wraps(fn)
189 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
190 try:
191 parameters = get_call_parameters(fn, args, kwargs, apply_defaults=False)
192 except Exception:
193 # Avoid raising any parsing exceptions here
194 parameters = kwargs
196 if name in parameters and when(parameters[name]):
197 message = generate_deprecation_message(
198 name=f"The parameter {name!r} for {fn.__name__!r}",
199 start_date=start_date,
200 end_date=end_date,
201 help=help,
202 when=when_message,
203 )
204 warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)
206 return fn(*args, **kwargs)
208 return wrapper
210 return decorator
213JsonValue: TypeAlias = Union[int, float, str, bool, None, list["JsonValue"], "JsonDict"] 1a
214JsonDict: TypeAlias = dict[str, JsonValue] 1a
217def deprecated_field( 1a
218 name: str,
219 *,
220 start_date: Optional[_AcceptableDate] = None,
221 end_date: Optional[_AcceptableDate] = None,
222 when_message: str = "",
223 help: str = "",
224 when: Optional[Callable[[Any], bool]] = None,
225 stacklevel: int = 2,
226) -> Callable[[type[M]], type[M]]:
227 """
228 Mark a field in a Pydantic model as deprecated.
230 Raises warning only if the field is specified during init.
232 Example:
234 ```python
236 @deprecated_field("x", when=lambda x: x is not None)
237 class Model(BaseModel)
238 x: Optional[int] = None
239 y: str
240 ```
241 """
243 when = when or (lambda _: True)
245 # Replaces the model's __init__ method with one that performs an additional warning
246 # check
247 def decorator(model_cls: type[M]) -> type[M]:
248 cls_init = model_cls.__init__
250 @functools.wraps(model_cls.__init__)
251 def __init__(__pydantic_self__: M, **data: Any) -> None:
252 if name in data.keys() and when(data[name]):
253 message = generate_deprecation_message(
254 name=f"The field {name!r} in {model_cls.__name__!r}",
255 start_date=start_date,
256 end_date=end_date,
257 help=help,
258 when=when_message,
259 )
260 warnings.warn(message, PrefectDeprecationWarning, stacklevel=stacklevel)
262 cls_init(__pydantic_self__, **data)
264 field = __pydantic_self__.__class__.model_fields.get(name)
265 if field is not None:
266 json_schema_extra = field.json_schema_extra or {}
268 if not isinstance(json_schema_extra, dict):
269 # json_schema_extra is a hook function; wrap it to add the deprecated flag.
270 extra_func = json_schema_extra
272 @functools.wraps(extra_func)
273 def wrapped(__json_schema: JsonDict) -> None:
274 extra_func(__json_schema)
275 __json_schema["deprecated"] = True
277 json_schema_extra = wrapped
279 else:
280 json_schema_extra["deprecated"] = True
282 field.json_schema_extra = json_schema_extra
284 # Patch the model's init method
285 model_cls.__init__ = __init__
287 return model_cls
289 return decorator
292def inject_renamed_module_alias_finder(): 1a
293 """
294 Insert an aliased module finder into Python's import machinery.
296 Required for `register_renamed_module` to work.
297 """
298 sys.meta_path.insert(0, AliasedModuleFinder(DEPRECATED_MODULE_ALIASES)) 1a
301def register_renamed_module(old_name: str, new_name: str, start_date: _AcceptableDate): 1a
302 """
303 Register a renamed module.
305 Adds backwwards compatibility imports for the old module name and displays a
306 deprecation warnings on import of the module.
307 """
308 message = generate_deprecation_message(
309 name=f"The {old_name!r} module",
310 start_date=start_date,
311 help=f"Use {new_name!r} instead.",
312 )
314 # Executed on module load
315 def callback(_):
316 return warnings.warn(message, DeprecationWarning, stacklevel=3)
318 DEPRECATED_MODULE_ALIASES.append(
319 AliasedModuleDefinition(old_name, new_name, callback)
320 )