Coverage for /usr/local/lib/python3.12/site-packages/prefect/assets/materialize.py: 50%

1from __future__ import annotations 1 ctx1a

2 

3from typing import TYPE_CHECKING, Callable, TypeVar, Union 1 ctx1a

4 

5from typing_extensions import ParamSpec, Unpack 1 ctx1a

6 

7from .core import Asset 1 ctx1a

8 

9T = TypeVar("T") 1 ctx1a

10P = ParamSpec("P") 1 ctx1a

11R = TypeVar("R") 1 ctx1a

12 

13if TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true1 ctx1a

14 from prefect.tasks import MaterializingTask, TaskOptions 

15 

16 

17def materialize( 1 ctx1a

18 *assets: Union[str, Asset], 

19 by: str | None = None, 

20 **task_kwargs: Unpack[TaskOptions], 

21) -> Callable[[Callable[P, R]], MaterializingTask[P, R]]: 

22 """ 

23 Decorator for materializing assets. 

24 

25 Args: 

26 *assets: Assets to materialize 

27 by: An optional tool that is ultimately responsible for materializing the asset e.g. "dbt" or "spark" 

28 **task_kwargs: Additional task configuration 

29 """ 

30 if not assets: 

31 raise TypeError( 

32 "materialize requires at least one asset argument, e.g. `@materialize(asset)`" 

33 ) 

34 

35 from prefect.tasks import MaterializingTask 

36 

37 def decorator(fn: Callable[P, R]) -> MaterializingTask[P, R]: 

38 return MaterializingTask( 

39 fn=fn, assets=assets, materialized_by=by, **task_kwargs 

40 ) 

41 

42 return decorator