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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
import uuid
from abc import abstractmethod
from os import PathLike
from typing import IO, Any, AnyStr, Dict, Optional, Union
from azure.ai.ml._exception_helper import log_and_raise_error
from azure.ai.ml._utils.utils import dump_yaml_to_file
from azure.ai.ml.entities._resource import Resource
from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException
class Asset(Resource):
"""Base class for asset.
This class should not be instantiated directly. Instead, use one of its subclasses.
:param name: The name of the asset. Defaults to a random GUID.
:type name: Optional[str]]
:param version: The version of the asset. Defaults to "1" if no name is provided, otherwise defaults to
autoincrement from the last registered version of the asset with that name. For a model name that has
never been registered, a default version will be assigned.
:type version: Optional[str]
:param description: The description of the resource. Defaults to None.
:type description: Optional[str]
:param tags: Tag dictionary. Tags can be added, removed, and updated. Defaults to None.
:type tags: Optional[dict[str, str]]
:param properties: The asset property dictionary. Defaults to None.
:type properties: Optional[dict[str, str]]
:keyword kwargs: A dictionary of additional configuration parameters.
:paramtype kwargs: Optional[dict]
"""
def __init__(
self,
name: Optional[str] = None,
version: Optional[str] = None,
description: Optional[str] = None,
tags: Optional[Dict] = None,
properties: Optional[Dict] = None,
**kwargs: Any,
) -> None:
self._is_anonymous = kwargs.pop("is_anonymous", False)
self._auto_increment_version = kwargs.pop("auto_increment_version", False)
self.auto_delete_setting = kwargs.pop("auto_delete_setting", None)
if not name and version is None:
name = _get_random_name()
version = "1"
self._is_anonymous = True
elif version is not None and not name:
msg = "If version is specified, name must be specified also."
err = ValidationException(
message=msg,
target=ErrorTarget.ASSET,
no_personal_data_message=msg,
error_category=ErrorCategory.USER_ERROR,
error_type=ValidationErrorType.MISSING_FIELD,
)
log_and_raise_error(err)
super().__init__(
name=name,
description=description,
tags=tags,
properties=properties,
**kwargs,
)
self.version = version
self.latest_version = None
@abstractmethod
def _to_dict(self) -> Dict:
"""Dump the artifact content into a pure dict object."""
@property
def version(self) -> Optional[str]:
"""The asset version.
:return: The asset version.
:rtype: str
"""
return self._version
@version.setter
def version(self, value: str) -> None:
"""Sets the asset version.
:param value: The asset version.
:type value: str
:raises ValidationException: Raised if value is not a string.
"""
if value:
if not isinstance(value, str):
msg = f"Asset version must be a string, not type {type(value)}."
err = ValidationException(
message=msg,
target=ErrorTarget.ASSET,
no_personal_data_message=msg,
error_category=ErrorCategory.USER_ERROR,
error_type=ValidationErrorType.INVALID_VALUE,
)
log_and_raise_error(err)
self._version = value
self._auto_increment_version = self.name and not self._version
def dump(self, dest: Union[str, PathLike, IO[AnyStr]], **kwargs: Any) -> None:
"""Dump the asset content into a file in YAML format.
:param dest: The local path or file stream to write the YAML content to.
If dest is a file path, a new file will be created.
If dest is an open file, the file will be written to directly.
:type dest: Union[PathLike, str, IO[AnyStr]]
:raises FileExistsError: Raised if dest is a file path and the file already exists.
:raises IOError: Raised if dest is an open file and the file is not writable.
"""
path = kwargs.pop("path", None)
yaml_serialized = self._to_dict()
dump_yaml_to_file(dest, yaml_serialized, default_flow_style=False, path=path, **kwargs)
def __eq__(self, other: Any) -> bool:
return bool(
self.name == other.name
and self.id == other.id
and self.version == other.version
and self.description == other.description
and self.tags == other.tags
and self.properties == other.properties
and self.base_path == other.base_path
and self._is_anonymous == other._is_anonymous
and self._auto_increment_version == other._auto_increment_version
and self.auto_delete_setting == other.auto_delete_setting
)
def __ne__(self, other: Any) -> bool:
return not self.__eq__(other)
def _get_random_name() -> str:
return str(uuid.uuid4())
|
