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

33 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 10:48 +0000

1from __future__ import annotations 1a

2 

3from typing import Any, ClassVar, Optional 1a

4 

5from pydantic import ConfigDict, Field 1a

6 

7from prefect._internal.schemas.bases import PrefectBaseModel 1a

8from prefect.types import ValidAssetKey 1a

9 

10MAX_ASSET_DESCRIPTION_LENGTH = 2500 1a

11 

12 

13class AssetProperties(PrefectBaseModel): 1a

14 """ 

15 Metadata properties to configure on an Asset 

16 """ 

17 

18 model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True) 1a

19 

20 name: Optional[str] = Field( 1a

21 default=None, description="Human readable name of the Asset." 

22 ) 

23 url: Optional[str] = Field( 1a

24 default=None, description="Visitable url to view the Asset." 

25 ) 

26 description: Optional[str] = Field( 1a

27 default=None, 

28 description="Description of the Asset.", 

29 max_length=MAX_ASSET_DESCRIPTION_LENGTH, 

30 ) 

31 owners: Optional[list[str]] = Field( 1a

32 default=None, description="Owners of the Asset." 

33 ) 

34 

35 

36class Asset(PrefectBaseModel): 1a

37 """ 

38 Assets are objects that represent materialized data, 

39 providing a way to track lineage and dependencies. 

40 """ 

41 

42 model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True) 1a

43 

44 key: ValidAssetKey 1a

45 properties: Optional[AssetProperties] = Field( 1a

46 default=None, 

47 description="Properties of the asset. " 

48 "Setting this will overwrite properties of a known asset.", 

49 ) 

50 

51 def __repr__(self) -> str: 1a

52 return f"Asset(key={self.key!r})" 

53 

54 def __hash__(self) -> int: 1a

55 return hash(self.key) 

56 

57 def add_metadata(self, metadata: dict[str, Any]) -> None: 1a

58 from prefect.context import AssetContext 

59 

60 asset_ctx = AssetContext.get() 

61 if not asset_ctx: 

62 raise RuntimeError( 

63 "Unable add Asset metadata when not inside of an AssetContext" 

64 ) 

65 

66 asset_ctx.add_asset_metadata(self.key, metadata) 

67 

68 

69def add_asset_metadata(asset: str | Asset, metadata: dict[str, Any]) -> None: 1a

70 from prefect.context import AssetContext 

71 

72 asset_ctx = AssetContext.get() 

73 if not asset_ctx: 

74 raise RuntimeError( 

75 "Unable to call `add_asset_metadata` when not inside of an AssetContext" 

76 ) 

77 

78 asset_key = asset if isinstance(asset, str) else asset.key 

79 asset_ctx.add_asset_metadata(asset_key, metadata)