1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
from os import PathLike
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from marshmallow import INCLUDE
from azure.ai.ml._schema._data.mltable_metadata_schema import MLTableMetadataSchema
from azure.ai.ml._utils.utils import load_yaml
from azure.ai.ml.constants._common import BASE_PATH_CONTEXT_KEY
from azure.ai.ml.entities._util import load_from_dict
class MLTableMetadataPath:
type: str # Literal["pattern", "file", "folder"]
value: Optional[str]
def __init__(self, *, pathDict: Dict):
if pathDict.get("pattern", None):
self.type = "pattern"
self.value = pathDict.get("pattern")
if pathDict.get("file", None):
self.type = "file"
self.value = pathDict.get("file")
if pathDict.get("folder", None):
self.type = "folder"
self.value = pathDict.get("folder")
class MLTableMetadata:
"""MLTableMetadata for data assets.
:param paths: List of paths which the MLTableMetadata refers to.
:type paths: List[MLTableMetadataPath]
:param transformations: Any transformations to be applied to the data referenced in paths.
:type transformations: List[Any]
:param base_path: Base path to resolve relative paths from.
:type base_path: str
"""
def __init__(
self,
*,
paths: List[MLTableMetadataPath],
transformations: Optional[List[Any]] = None,
base_path: str,
**_kwargs: Any,
):
self.base_path = base_path
self.paths = paths
self.transformations = transformations
@classmethod
def load(
cls,
yaml_path: Union[PathLike, str],
**kwargs: Any,
) -> "MLTableMetadata":
"""Construct an MLTable object from yaml file.
:param yaml_path: Path to a local file as the source.
:type yaml_path: PathLike | str
:return: Constructed MLTable object.
:rtype: MLTable
"""
yaml_dict = load_yaml(yaml_path)
return cls._load(yaml_data=yaml_dict, yaml_path=yaml_path, **kwargs)
@classmethod
def _load(
cls,
yaml_data: Optional[Dict],
yaml_path: Optional[Union[PathLike, str]],
**kwargs: Any,
) -> "MLTableMetadata":
yaml_data = yaml_data or {}
context = {
BASE_PATH_CONTEXT_KEY: Path(yaml_path).parent if yaml_path else Path("./"),
}
res: MLTableMetadata = load_from_dict(MLTableMetadataSchema, yaml_data, context, "", unknown=INCLUDE, **kwargs)
return res
def _to_dict(self) -> Dict:
res: dict = MLTableMetadataSchema(context={BASE_PATH_CONTEXT_KEY: "./"}, unknown=INCLUDE).dump(self)
return res
def referenced_uris(self) -> List[Optional[str]]:
return [path.value for path in self.paths]
|
