Skip to content

Artifact

API Classes

Found a method here you want to use? The API Classes have method pass-through so just call the method on the any class that inherits from the Artifact Class and voilà it works the same.

The Workbench Artifact class is a base/abstract class that defines API implemented by all the child classes (DataSource, FeatureSet, Model, Endpoint).

Artifact is storage-agnostic: naming rules, the abstract method set, and every helper expressible in terms of workbench_meta()/upsert_workbench_meta() — tags, owner, input, status, health. Those two metadata primitives are abstract, and two classes provide them:

  • AWSArtifact backs metadata with AWS tags and carries the shared AWS session and bucket paths. Every AWS artifact class inherits from it.
  • LocalArtifact backs metadata with an on-disk meta.json and carries no AWS session, bucket, or ARN.

arn(), aws_url(), and aws_meta() are declared on AWSArtifact rather than on the base, since local artifacts have no such thing.

Artifact: Abstract Base Class for all Artifact classes in Workbench.

Artifact is the storage-agnostic contract: naming rules, the abstract method set, and every helper that can be expressed in terms of workbench_meta()/upsert_workbench_meta() (tags, owner, input, status, health). AWSArtifact backs that metadata with AWS tags and provides the shared AWS session/bucket resources; workbench.local.LocalArtifact backs it with an on-disk meta.json.

Artifact

Bases: ABC

Artifact: Abstract Base Class for all Artifact classes in Workbench

Source code in src/workbench/core/artifact.py 
 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
class Artifact(ABC):
    """Artifact: Abstract Base Class for all Artifact classes in Workbench"""

    # Class-level shared resources
    log = logging.getLogger("workbench")

    # Delimiter for storing lists in metadata
    tag_delimiter = "::"

    def __init__(self, name: str, **kwargs):
        """Initialize the Artifact Base Class

        Args:
            name (str): The Name of this artifact
        """
        self.name = name

    def __post_init__(self):
        """Artifact Post Initialization"""

        # Do I exist? (very metaphysical)
        if not self.exists():
            self.log.debug(f"Artifact {self.name} does not exist")
            return

        # Conduct a Health Check on this Artifact
        health_issues = self.health_check()
        if health_issues:
            if "needs_onboard" in health_issues:
                self.log.important(f"Artifact {self.name} needs to be onboarded")
            elif health_issues == ["no_activity"]:
                self.log.debug(f"Artifact {self.name} has no activity, which is fine")
            else:
                self.log.warning(f"Health Check Failed {self.name}: {health_issues}")
            for issue in health_issues:
                self.add_health_tag(issue)
        else:
            self.log.info(f"Health Check Passed {self.name}")

    @classmethod
    def is_name_valid(cls, name: str, delimiter: str = "_", lower_case: bool = True) -> bool:
        """Check if the name adheres to the naming conventions for this Artifact.

        Args:
            name (str): The name/id to check.
            delimiter (str): The delimiter to use in the name/id string (default: "_")
            lower_case (bool): Should the name be lowercased? (default: True)

        Returns:
            bool: True if the name is valid, False otherwise.
        """
        valid_name = cls.generate_valid_name(name, delimiter=delimiter, lower_case=lower_case)
        if name != valid_name:
            cls.log.warning(f"Artifact name: '{name}' is not valid. Convert it to something like: '{valid_name}'")
            return False
        return True

    @staticmethod
    def generate_valid_name(name: str, delimiter: str = "_", lower_case: bool = True) -> str:
        """Only allow letters and the specified delimiter, also lowercase the string.

        Args:
            name (str): The name/id string to check.
            delimiter (str): The delimiter to use in the name/id string (default: "_")
            lower_case (bool): Should the name be lowercased? (default: True)

        Returns:
            str: A generated valid name/id.
        """
        valid_name = "".join(c for c in name if c.isalnum() or c in ["_", "-"])
        if lower_case:
            valid_name = valid_name.lower()

        # Replace with the chosen delimiter
        return valid_name.replace("_", delimiter).replace("-", delimiter)

    @abstractmethod
    def exists(self) -> bool:
        """Does the Artifact exist? Can we connect to it?"""
        pass

    @abstractmethod
    def workbench_meta(self) -> Union[dict, None]:
        """Get the Workbench specific metadata for this Artifact

        Returns:
            Union[dict, None]: Dictionary of Workbench metadata for this Artifact
        """
        pass

    @abstractmethod
    def upsert_workbench_meta(self, new_meta: dict):
        """Add Workbench specific metadata to this Artifact

        Args:
            new_meta (dict): Dictionary of NEW metadata to add
        """
        pass

    def expected_meta(self) -> list[str]:
        """Metadata we expect to see for this Artifact when it's ready
        Returns:
            list[str]: List of expected metadata keys
        """

        # If an artifact has additional expected metadata override this method
        return ["workbench_status"]

    @abstractmethod
    def refresh_meta(self):
        """Refresh the Artifact's metadata"""
        pass

    def ready(self) -> bool:
        """Is the Artifact ready? Is initial setup complete and expected metadata populated?"""

        # If anything goes wrong, assume the artifact is not ready
        try:
            # Check for the expected metadata
            expected_meta = self.expected_meta()
            existing_meta = self.workbench_meta()
            ready = set(existing_meta.keys()).issuperset(expected_meta)
            if ready:
                return True
            else:
                self.log.info("Artifact is not ready!")
                return False
        except Exception as e:
            self.log.error(f"Artifact malformed: {e}")
            return False

    @abstractmethod
    def onboard(self) -> bool:
        """Onboard this Artifact into Workbench
        Returns:
            bool: True if the Artifact was successfully onboarded, False otherwise
        """
        pass

    @abstractmethod
    def details(self) -> dict:
        """Additional Details about this Artifact"""
        pass

    @abstractmethod
    def size(self) -> float:
        """Return the size of this artifact in MegaBytes"""
        pass

    @abstractmethod
    def created(self) -> datetime:
        """Return the datetime when this artifact was created"""
        pass

    @abstractmethod
    def modified(self) -> datetime:
        """Return the datetime when this artifact was last modified"""
        pass

    @abstractmethod
    def hash(self) -> str:
        """Return the hash of this artifact, useful for content validation"""
        pass

    @abstractmethod
    def delete(self):
        """Delete this artifact including all related objects"""
        pass

    def get_tags(self, tag_type="user") -> list:
        """Get the tags for this artifact
        Args:
            tag_type (str): Type of tags to return (user or health)
        Returns:
            list[str]: List of tags for this artifact
        """
        if tag_type == "user":
            user_tags = self.workbench_meta().get("workbench_tags")
            return user_tags.split(self.tag_delimiter) if user_tags else []

        # Grab our health tags
        health_tags = self.workbench_meta().get("workbench_health_tags")

        # If we don't have health tags, create the storage and return an empty list
        if health_tags is None:
            self.log.important(f"{self.name} creating workbench_health_tags storage...")
            self.upsert_workbench_meta({"workbench_health_tags": ""})
            return []

        # Otherwise, return the health tags
        return health_tags.split(self.tag_delimiter) if health_tags else []

    def set_tags(self, tags):
        self.upsert_workbench_meta({"workbench_tags": self.tag_delimiter.join(tags)})

    def add_tag(self, tag, tag_type="user"):
        """Add a tag for this artifact, ensuring no duplicates and maintaining order.
        Args:
            tag (str): Tag to add for this artifact
            tag_type (str): Type of tag to add (user or health)
        """
        current_tags = self.get_tags(tag_type) if tag_type == "user" else self.get_health_tags()
        if tag not in current_tags:
            current_tags.append(tag)
            combined_tags = self.tag_delimiter.join(current_tags)
            if tag_type == "user":
                self.upsert_workbench_meta({"workbench_tags": combined_tags})
            else:
                self.upsert_workbench_meta({"workbench_health_tags": combined_tags})

    def remove_workbench_tag(self, tag, tag_type="user"):
        """Remove a tag from this artifact if it exists.
        Args:
            tag (str): Tag to remove from this artifact
            tag_type (str): Type of tag to remove (user or health)
        """
        current_tags = self.get_tags(tag_type) if tag_type == "user" else self.get_health_tags()
        if tag in current_tags:
            current_tags.remove(tag)
            combined_tags = self.tag_delimiter.join(current_tags)
            if tag_type == "user":
                self.upsert_workbench_meta({"workbench_tags": combined_tags})
            elif tag_type == "health":
                self.upsert_workbench_meta({"workbench_health_tags": combined_tags})

    # Syntactic sugar for health tags
    def get_health_tags(self):
        return self.get_tags(tag_type="health")

    def set_health_tags(self, tags):
        self.upsert_workbench_meta({"workbench_health_tags": self.tag_delimiter.join(tags)})

    def add_health_tag(self, tag):
        self.add_tag(tag, tag_type="health")

    def remove_health_tag(self, tag):
        self.remove_workbench_tag(tag, tag_type="health")

    # Owner of this artifact
    def get_owner(self) -> str:
        """Get the owner of this artifact"""
        return self.workbench_meta().get("workbench_owner", "unknown")

    def set_owner(self, owner: str):
        """Set the owner of this artifact

        Args:
            owner (str): Owner to set for this artifact
        """
        self.upsert_workbench_meta({"workbench_owner": owner})

    def get_input(self) -> str:
        """Get the input data for this artifact"""
        return self.workbench_meta().get("workbench_input", "unknown")

    def get_status(self) -> str:
        """Get the status for this artifact"""
        return self.workbench_meta().get("workbench_status", "unknown")

    def set_status(self, status: str):
        """Set the status for this artifact
        Args:
            status (str): Status to set for this artifact
        """
        self.upsert_workbench_meta({"workbench_status": status})

    def health_check(self, deep: bool = False) -> list[str]:
        """Perform a health check on this artifact

        Args:
            deep (bool): If True, perform more extensive (expensive) health checks (default: False)

        Returns:
            list[str]: List of health issues
        """
        health_issues = []
        if not self.ready():
            return ["needs_onboard"]
        # FIXME: Revisit AWS URL check ("unknown" in aws_url() -> "aws_url_unknown" health issue)
        return health_issues

    def summary(self) -> dict:
        """This is generic summary information for all Artifacts. If you
        want to get more detailed information, call the details() method
        which is implemented by the specific Artifact class"""
        basic = {
            "name": self.name,
            "health_tags": self.get_health_tags(),
            "size": self.size(),
            "created": self.created(),
            "modified": self.modified(),
            "input": self.get_input(),
        }
        # Combine the workbench metadata with the basic metadata
        return {**basic, **self.workbench_meta()}

    def __repr__(self) -> str:
        """String representation of this artifact

        Returns:
            str: String representation of this artifact
        """

        # If the artifact does not exist, return a message
        if not self.exists():
            return f"{self.__class__.__name__}: {self.name} does not exist"

        summary_dict = self.summary()
        display_keys = [
            "aws_arn",
            "health_tags",
            "size",
            "created",
            "modified",
            "input",
            "workbench_status",
            "workbench_tags",
        ]
        summary_items = [f"  {repr(key)}: {repr(value)}" for key, value in summary_dict.items() if key in display_keys]
        summary_str = f"{self.__class__.__name__}: {self.name}\n" + ",\n".join(summary_items)
        return summary_str

__init__(name, **kwargs)

Initialize the Artifact Base Class

Parameters:

Name Type Description Default
name str

The Name of this artifact

 required
Source code in src/workbench/core/artifact.py 
26
27
28
29
30
31
32
def __init__(self, name: str, **kwargs):
    """Initialize the Artifact Base Class

    Args:
        name (str): The Name of this artifact
    """
    self.name = name

__post_init__()

Artifact Post Initialization

Source code in src/workbench/core/artifact.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __post_init__(self):
    """Artifact Post Initialization"""

    # Do I exist? (very metaphysical)
    if not self.exists():
        self.log.debug(f"Artifact {self.name} does not exist")
        return

    # Conduct a Health Check on this Artifact
    health_issues = self.health_check()
    if health_issues:
        if "needs_onboard" in health_issues:
            self.log.important(f"Artifact {self.name} needs to be onboarded")
        elif health_issues == ["no_activity"]:
            self.log.debug(f"Artifact {self.name} has no activity, which is fine")
        else:
            self.log.warning(f"Health Check Failed {self.name}: {health_issues}")
        for issue in health_issues:
            self.add_health_tag(issue)
    else:
        self.log.info(f"Health Check Passed {self.name}")

__repr__()

String representation of this artifact

Returns:

Name Type Description
str str

String representation of this artifact
Source code in src/workbench/core/artifact.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def __repr__(self) -> str:
    """String representation of this artifact

    Returns:
        str: String representation of this artifact
    """

    # If the artifact does not exist, return a message
    if not self.exists():
        return f"{self.__class__.__name__}: {self.name} does not exist"

    summary_dict = self.summary()
    display_keys = [
        "aws_arn",
        "health_tags",
        "size",
        "created",
        "modified",
        "input",
        "workbench_status",
        "workbench_tags",
    ]
    summary_items = [f"  {repr(key)}: {repr(value)}" for key, value in summary_dict.items() if key in display_keys]
    summary_str = f"{self.__class__.__name__}: {self.name}\n" + ",\n".join(summary_items)
    return summary_str

add_tag(tag, tag_type='user')

Add a tag for this artifact, ensuring no duplicates and maintaining order. Args: tag (str): Tag to add for this artifact tag_type (str): Type of tag to add (user or health)

Source code in src/workbench/core/artifact.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def add_tag(self, tag, tag_type="user"):
    """Add a tag for this artifact, ensuring no duplicates and maintaining order.
    Args:
        tag (str): Tag to add for this artifact
        tag_type (str): Type of tag to add (user or health)
    """
    current_tags = self.get_tags(tag_type) if tag_type == "user" else self.get_health_tags()
    if tag not in current_tags:
        current_tags.append(tag)
        combined_tags = self.tag_delimiter.join(current_tags)
        if tag_type == "user":
            self.upsert_workbench_meta({"workbench_tags": combined_tags})
        else:
            self.upsert_workbench_meta({"workbench_health_tags": combined_tags})

created() abstractmethod

Return the datetime when this artifact was created

Source code in src/workbench/core/artifact.py 
166
167
168
169
@abstractmethod
def created(self) -> datetime:
    """Return the datetime when this artifact was created"""
    pass

delete() abstractmethod

Delete this artifact including all related objects

Source code in src/workbench/core/artifact.py 
181
182
183
184
@abstractmethod
def delete(self):
    """Delete this artifact including all related objects"""
    pass

details() abstractmethod

Additional Details about this Artifact

Source code in src/workbench/core/artifact.py 
156
157
158
159
@abstractmethod
def details(self) -> dict:
    """Additional Details about this Artifact"""
    pass

exists() abstractmethod

Does the Artifact exist? Can we connect to it?

Source code in src/workbench/core/artifact.py 
93
94
95
96
@abstractmethod
def exists(self) -> bool:
    """Does the Artifact exist? Can we connect to it?"""
    pass

expected_meta()

Metadata we expect to see for this Artifact when it's ready Returns: list[str]: List of expected metadata keys

Source code in src/workbench/core/artifact.py 
116
117
118
119
120
121
122
123
def expected_meta(self) -> list[str]:
    """Metadata we expect to see for this Artifact when it's ready
    Returns:
        list[str]: List of expected metadata keys
    """

    # If an artifact has additional expected metadata override this method
    return ["workbench_status"]

generate_valid_name(name, delimiter='_', lower_case=True) staticmethod

Only allow letters and the specified delimiter, also lowercase the string.

Parameters:

Name Type Description Default
name str

The name/id string to check.

 required
delimiter str

The delimiter to use in the name/id string (default: "_")

 '_'
lower_case bool

Should the name be lowercased? (default: True)

 True

Returns:

Name Type Description
str str

A generated valid name/id.
Source code in src/workbench/core/artifact.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@staticmethod
def generate_valid_name(name: str, delimiter: str = "_", lower_case: bool = True) -> str:
    """Only allow letters and the specified delimiter, also lowercase the string.

    Args:
        name (str): The name/id string to check.
        delimiter (str): The delimiter to use in the name/id string (default: "_")
        lower_case (bool): Should the name be lowercased? (default: True)

    Returns:
        str: A generated valid name/id.
    """
    valid_name = "".join(c for c in name if c.isalnum() or c in ["_", "-"])
    if lower_case:
        valid_name = valid_name.lower()

    # Replace with the chosen delimiter
    return valid_name.replace("_", delimiter).replace("-", delimiter)

get_input()

Get the input data for this artifact

Source code in src/workbench/core/artifact.py 
268
269
270
def get_input(self) -> str:
    """Get the input data for this artifact"""
    return self.workbench_meta().get("workbench_input", "unknown")

get_owner()

Get the owner of this artifact

Source code in src/workbench/core/artifact.py 
256
257
258
def get_owner(self) -> str:
    """Get the owner of this artifact"""
    return self.workbench_meta().get("workbench_owner", "unknown")

get_status()

Get the status for this artifact

Source code in src/workbench/core/artifact.py 
272
273
274
def get_status(self) -> str:
    """Get the status for this artifact"""
    return self.workbench_meta().get("workbench_status", "unknown")

get_tags(tag_type='user')

Get the tags for this artifact Args: tag_type (str): Type of tags to return (user or health) Returns: list[str]: List of tags for this artifact

Source code in src/workbench/core/artifact.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def get_tags(self, tag_type="user") -> list:
    """Get the tags for this artifact
    Args:
        tag_type (str): Type of tags to return (user or health)
    Returns:
        list[str]: List of tags for this artifact
    """
    if tag_type == "user":
        user_tags = self.workbench_meta().get("workbench_tags")
        return user_tags.split(self.tag_delimiter) if user_tags else []

    # Grab our health tags
    health_tags = self.workbench_meta().get("workbench_health_tags")

    # If we don't have health tags, create the storage and return an empty list
    if health_tags is None:
        self.log.important(f"{self.name} creating workbench_health_tags storage...")
        self.upsert_workbench_meta({"workbench_health_tags": ""})
        return []

    # Otherwise, return the health tags
    return health_tags.split(self.tag_delimiter) if health_tags else []

hash() abstractmethod

Return the hash of this artifact, useful for content validation

Source code in src/workbench/core/artifact.py 
176
177
178
179
@abstractmethod
def hash(self) -> str:
    """Return the hash of this artifact, useful for content validation"""
    pass

health_check(deep=False)

Perform a health check on this artifact

Parameters:

Name Type Description Default
deep bool

If True, perform more extensive (expensive) health checks (default: False)

 False

Returns:

Type Description
list[str]

list[str]: List of health issues
Source code in src/workbench/core/artifact.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def health_check(self, deep: bool = False) -> list[str]:
    """Perform a health check on this artifact

    Args:
        deep (bool): If True, perform more extensive (expensive) health checks (default: False)

    Returns:
        list[str]: List of health issues
    """
    health_issues = []
    if not self.ready():
        return ["needs_onboard"]
    # FIXME: Revisit AWS URL check ("unknown" in aws_url() -> "aws_url_unknown" health issue)
    return health_issues

is_name_valid(name, delimiter='_', lower_case=True) classmethod

Check if the name adheres to the naming conventions for this Artifact.

Parameters:

Name Type Description Default
name str

The name/id to check.

 required
delimiter str

The delimiter to use in the name/id string (default: "_")

 '_'
lower_case bool

Should the name be lowercased? (default: True)

 True

Returns:

Name Type Description
bool bool

True if the name is valid, False otherwise.
Source code in src/workbench/core/artifact.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def is_name_valid(cls, name: str, delimiter: str = "_", lower_case: bool = True) -> bool:
    """Check if the name adheres to the naming conventions for this Artifact.

    Args:
        name (str): The name/id to check.
        delimiter (str): The delimiter to use in the name/id string (default: "_")
        lower_case (bool): Should the name be lowercased? (default: True)

    Returns:
        bool: True if the name is valid, False otherwise.
    """
    valid_name = cls.generate_valid_name(name, delimiter=delimiter, lower_case=lower_case)
    if name != valid_name:
        cls.log.warning(f"Artifact name: '{name}' is not valid. Convert it to something like: '{valid_name}'")
        return False
    return True

modified() abstractmethod

Return the datetime when this artifact was last modified

Source code in src/workbench/core/artifact.py 
171
172
173
174
@abstractmethod
def modified(self) -> datetime:
    """Return the datetime when this artifact was last modified"""
    pass

onboard() abstractmethod

Onboard this Artifact into Workbench Returns: bool: True if the Artifact was successfully onboarded, False otherwise

Source code in src/workbench/core/artifact.py 
148
149
150
151
152
153
154
@abstractmethod
def onboard(self) -> bool:
    """Onboard this Artifact into Workbench
    Returns:
        bool: True if the Artifact was successfully onboarded, False otherwise
    """
    pass

ready()

Is the Artifact ready? Is initial setup complete and expected metadata populated?

Source code in src/workbench/core/artifact.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def ready(self) -> bool:
    """Is the Artifact ready? Is initial setup complete and expected metadata populated?"""

    # If anything goes wrong, assume the artifact is not ready
    try:
        # Check for the expected metadata
        expected_meta = self.expected_meta()
        existing_meta = self.workbench_meta()
        ready = set(existing_meta.keys()).issuperset(expected_meta)
        if ready:
            return True
        else:
            self.log.info("Artifact is not ready!")
            return False
    except Exception as e:
        self.log.error(f"Artifact malformed: {e}")
        return False

refresh_meta() abstractmethod

Refresh the Artifact's metadata

Source code in src/workbench/core/artifact.py 
125
126
127
128
@abstractmethod
def refresh_meta(self):
    """Refresh the Artifact's metadata"""
    pass

remove_workbench_tag(tag, tag_type='user')

Remove a tag from this artifact if it exists. Args: tag (str): Tag to remove from this artifact tag_type (str): Type of tag to remove (user or health)

Source code in src/workbench/core/artifact.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def remove_workbench_tag(self, tag, tag_type="user"):
    """Remove a tag from this artifact if it exists.
    Args:
        tag (str): Tag to remove from this artifact
        tag_type (str): Type of tag to remove (user or health)
    """
    current_tags = self.get_tags(tag_type) if tag_type == "user" else self.get_health_tags()
    if tag in current_tags:
        current_tags.remove(tag)
        combined_tags = self.tag_delimiter.join(current_tags)
        if tag_type == "user":
            self.upsert_workbench_meta({"workbench_tags": combined_tags})
        elif tag_type == "health":
            self.upsert_workbench_meta({"workbench_health_tags": combined_tags})

set_owner(owner)

Set the owner of this artifact

Parameters:

Name Type Description Default
owner str

Owner to set for this artifact

 required
Source code in src/workbench/core/artifact.py 
260
261
262
263
264
265
266
def set_owner(self, owner: str):
    """Set the owner of this artifact

    Args:
        owner (str): Owner to set for this artifact
    """
    self.upsert_workbench_meta({"workbench_owner": owner})

set_status(status)

Set the status for this artifact Args: status (str): Status to set for this artifact

Source code in src/workbench/core/artifact.py 
276
277
278
279
280
281
def set_status(self, status: str):
    """Set the status for this artifact
    Args:
        status (str): Status to set for this artifact
    """
    self.upsert_workbench_meta({"workbench_status": status})

size() abstractmethod

Return the size of this artifact in MegaBytes

Source code in src/workbench/core/artifact.py 
161
162
163
164
@abstractmethod
def size(self) -> float:
    """Return the size of this artifact in MegaBytes"""
    pass

summary()

This is generic summary information for all Artifacts. If you want to get more detailed information, call the details() method which is implemented by the specific Artifact class

Source code in src/workbench/core/artifact.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def summary(self) -> dict:
    """This is generic summary information for all Artifacts. If you
    want to get more detailed information, call the details() method
    which is implemented by the specific Artifact class"""
    basic = {
        "name": self.name,
        "health_tags": self.get_health_tags(),
        "size": self.size(),
        "created": self.created(),
        "modified": self.modified(),
        "input": self.get_input(),
    }
    # Combine the workbench metadata with the basic metadata
    return {**basic, **self.workbench_meta()}

upsert_workbench_meta(new_meta) abstractmethod

Add Workbench specific metadata to this Artifact

Parameters:

Name Type Description Default
new_meta dict

Dictionary of NEW metadata to add

 required
Source code in src/workbench/core/artifact.py 
107
108
109
110
111
112
113
114
@abstractmethod
def upsert_workbench_meta(self, new_meta: dict):
    """Add Workbench specific metadata to this Artifact

    Args:
        new_meta (dict): Dictionary of NEW metadata to add
    """
    pass

workbench_meta() abstractmethod

Get the Workbench specific metadata for this Artifact

Returns:

Type Description
Union[dict, None]

Union[dict, None]: Dictionary of Workbench metadata for this Artifact
Source code in src/workbench/core/artifact.py 
 98
 99
100
101
102
103
104
105
@abstractmethod
def workbench_meta(self) -> Union[dict, None]:
    """Get the Workbench specific metadata for this Artifact

    Returns:
        Union[dict, None]: Dictionary of Workbench metadata for this Artifact
    """
    pass

AWSArtifact: Base Class for all AWS-backed Artifact classes in Workbench.

Backs the Artifact metadata contract with AWS tags and provides the shared AWS session/bucket resources used by every AWS artifact class.

AWSArtifact

Bases: Artifact

AWSArtifact: Base Class for all AWS-backed Artifact classes in Workbench

Source code in src/workbench/core/artifacts/aws_artifact.py 
 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class AWSArtifact(Artifact):
    """AWSArtifact: Base Class for all AWS-backed Artifact classes in Workbench"""

    # Config Manager
    cm = ConfigManager()
    if not cm.config_okay():
        log = logging.getLogger("workbench")
        log.critical("Workbench Configuration Incomplete...")
        log.critical("Run the 'workbench' command and follow the prompts...")
        raise FatalConfigError()

    # AWS Account Clamp
    aws_account_clamp = AWSAccountClamp()
    boto3_session = aws_account_clamp.boto3_session
    sm_session = aws_account_clamp.sagemaker_session()
    sm_client = aws_account_clamp.sagemaker_client()
    aws_region = aws_account_clamp.region

    # Setup Bucket Paths
    workbench_bucket = cm.get_config("WORKBENCH_BUCKET")
    data_sources_s3_path = f"s3://{workbench_bucket}/data-sources"
    feature_sets_s3_path = f"s3://{workbench_bucket}/feature-sets"
    models_s3_path = f"s3://{workbench_bucket}/models"
    endpoints_s3_path = f"s3://{workbench_bucket}/endpoints"
    # Scratch root for transient files, separate from the protected artifact
    # prefixes. Each use owns a subfolder (temp/training_data/, temp/athena_output/).
    temp_s3_path = f"s3://{workbench_bucket}/temp"

    # Grab our Dataframe Cache Storage (use the endpoint-safe core class directly
    # with our refreshable session + config-loaded bucket — equivalent to going
    # through workbench.api.DFStore but without triggering workbench.api.__init__
    # while artifact.py is still loading).
    df_cache = DFStoreCore(
        path_prefix="/workbench/dataframe_cache",
        s3_bucket=workbench_bucket,
        boto3_session=boto3_session,
    )

    # Artifact may want to use the Parameter Store or Dataframe Store
    param_store = ParameterStore(boto3_session=boto3_session)
    df_store = DFStoreCore(s3_bucket=workbench_bucket, boto3_session=boto3_session)

    def __init__(self, name: str, **kwargs):
        """Initialize the AWSArtifact Base Class

        Args:
            name (str): The Name of this artifact
        """
        super().__init__(name, **kwargs)
        self.meta = CloudMeta()

    @abstractmethod
    def arn(self):
        """AWS ARN (Amazon Resource Name) for this artifact"""
        pass

    @abstractmethod
    def aws_url(self):
        """AWS console/web interface for this artifact"""
        pass

    @abstractmethod
    def aws_meta(self) -> dict:
        """Get the full AWS metadata for this artifact"""
        pass

    def summary(self) -> dict:
        """Generic summary information, plus the artifact's ARN"""
        return {**super().summary(), "aws_arn": self.arn()}

    def workbench_meta(self) -> Union[dict, None]:
        """Get the Workbench specific metadata for this Artifact

        Returns:
            Union[dict, None]: Dictionary of Workbench metadata for this Artifact

        Note: This functionality will work for FeatureSets, Models, and Endpoints
              but not for DataSources and Graphs, those classes need to override this method.
        """
        return self.meta.get_aws_tags(self.arn())

    @aws_throttle
    def upsert_workbench_meta(self, new_meta: dict):
        """Add Workbench specific metadata to this Artifact
        Args:
            new_meta (dict): Dictionary of NEW metadata to add
        Note:
            This functionality will work for FeatureSets, Models, and Endpoints
            but not for DataSources. The DataSource class overrides this method.
        """

        # Check for ReadOnly Role
        if self.aws_account_clamp.read_only:
            self.log.info("Cannot add metadata with a ReadOnly Permissions...")
            return

        # Sanity check
        aws_arn = self.arn()
        if aws_arn is None:
            self.log.error(f"ARN is None for {self.name}!")
            return

        # Add the new metadata to the existing metadata
        self.log.info(f"Adding Tags to {self.name}:{str(new_meta)[:50]}...")
        aws_tags = dict_to_aws_tags(new_meta)
        try:
            Tag.add_tags(resource_arn=aws_arn, tags=aws_tags, session=self.boto3_session)
        except ClientError as e:
            if e.response["Error"]["Code"] == "ThrottlingException":
                raise  # @aws_throttle handles the backoff/retry
            self.log.error(f"Error adding metadata to {aws_arn}: {type(e).__name__}: {e}")
            return
        except Exception as e:
            self.log.error(f"Error adding metadata to {aws_arn}: {type(e).__name__}: {e}")
            return

        # Poke the modified registry so caches know this artifact changed
        from workbench.cached.cached_meta import CachedMeta

        CachedMeta().update_modified_timestamp(self)

    @aws_throttle
    def delete_metadata(self, key_to_delete: str):
        """Delete specific metadata from this artifact
        Args:
            key_to_delete (str): Metadata key to delete
        """

        aws_arn = self.arn()
        self.log.important(f"Deleting Metadata {key_to_delete} for Artifact: {aws_arn}...")

        # First, fetch all the existing tags using V3 API
        from sagemaker.core.common_utils import list_tags as sm_list_tags

        existing_tags = sm_list_tags(self.sm_session, aws_arn)

        # Convert existing AWS tags to a dictionary for easy manipulation
        existing_tags_dict = {item["Key"]: item["Value"] for item in existing_tags}

        # Identify tags to delete
        tag_list_to_delete = []
        for key in existing_tags_dict.keys():
            if key == key_to_delete or key.startswith(f"{key_to_delete}_chunk_"):
                tag_list_to_delete.append(key)

        # Delete the identified tags using V3 API
        if tag_list_to_delete:
            Tag.delete_tags(resource_arn=aws_arn, tag_keys=tag_list_to_delete, session=self.boto3_session)
        else:
            self.log.info(f"No Metadata found: {key_to_delete}...")

__init__(name, **kwargs)

Initialize the AWSArtifact Base Class

Parameters:

Name Type Description Default
name str

The Name of this artifact

 required
Source code in src/workbench/core/artifacts/aws_artifact.py 
65
66
67
68
69
70
71
72
def __init__(self, name: str, **kwargs):
    """Initialize the AWSArtifact Base Class

    Args:
        name (str): The Name of this artifact
    """
    super().__init__(name, **kwargs)
    self.meta = CloudMeta()

arn() abstractmethod

AWS ARN (Amazon Resource Name) for this artifact

Source code in src/workbench/core/artifacts/aws_artifact.py 
74
75
76
77
@abstractmethod
def arn(self):
    """AWS ARN (Amazon Resource Name) for this artifact"""
    pass

aws_meta() abstractmethod

Get the full AWS metadata for this artifact

Source code in src/workbench/core/artifacts/aws_artifact.py 
84
85
86
87
@abstractmethod
def aws_meta(self) -> dict:
    """Get the full AWS metadata for this artifact"""
    pass

aws_url() abstractmethod

AWS console/web interface for this artifact

Source code in src/workbench/core/artifacts/aws_artifact.py 
79
80
81
82
@abstractmethod
def aws_url(self):
    """AWS console/web interface for this artifact"""
    pass

delete_metadata(key_to_delete)

Delete specific metadata from this artifact Args: key_to_delete (str): Metadata key to delete

Source code in src/workbench/core/artifacts/aws_artifact.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@aws_throttle
def delete_metadata(self, key_to_delete: str):
    """Delete specific metadata from this artifact
    Args:
        key_to_delete (str): Metadata key to delete
    """

    aws_arn = self.arn()
    self.log.important(f"Deleting Metadata {key_to_delete} for Artifact: {aws_arn}...")

    # First, fetch all the existing tags using V3 API
    from sagemaker.core.common_utils import list_tags as sm_list_tags

    existing_tags = sm_list_tags(self.sm_session, aws_arn)

    # Convert existing AWS tags to a dictionary for easy manipulation
    existing_tags_dict = {item["Key"]: item["Value"] for item in existing_tags}

    # Identify tags to delete
    tag_list_to_delete = []
    for key in existing_tags_dict.keys():
        if key == key_to_delete or key.startswith(f"{key_to_delete}_chunk_"):
            tag_list_to_delete.append(key)

    # Delete the identified tags using V3 API
    if tag_list_to_delete:
        Tag.delete_tags(resource_arn=aws_arn, tag_keys=tag_list_to_delete, session=self.boto3_session)
    else:
        self.log.info(f"No Metadata found: {key_to_delete}...")

summary()

Generic summary information, plus the artifact's ARN

Source code in src/workbench/core/artifacts/aws_artifact.py 
89
90
91
def summary(self) -> dict:
    """Generic summary information, plus the artifact's ARN"""
    return {**super().summary(), "aws_arn": self.arn()}

upsert_workbench_meta(new_meta)

Add Workbench specific metadata to this Artifact Args: new_meta (dict): Dictionary of NEW metadata to add Note: This functionality will work for FeatureSets, Models, and Endpoints but not for DataSources. The DataSource class overrides this method.

Source code in src/workbench/core/artifacts/aws_artifact.py 
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
@aws_throttle
def upsert_workbench_meta(self, new_meta: dict):
    """Add Workbench specific metadata to this Artifact
    Args:
        new_meta (dict): Dictionary of NEW metadata to add
    Note:
        This functionality will work for FeatureSets, Models, and Endpoints
        but not for DataSources. The DataSource class overrides this method.
    """

    # Check for ReadOnly Role
    if self.aws_account_clamp.read_only:
        self.log.info("Cannot add metadata with a ReadOnly Permissions...")
        return

    # Sanity check
    aws_arn = self.arn()
    if aws_arn is None:
        self.log.error(f"ARN is None for {self.name}!")
        return

    # Add the new metadata to the existing metadata
    self.log.info(f"Adding Tags to {self.name}:{str(new_meta)[:50]}...")
    aws_tags = dict_to_aws_tags(new_meta)
    try:
        Tag.add_tags(resource_arn=aws_arn, tags=aws_tags, session=self.boto3_session)
    except ClientError as e:
        if e.response["Error"]["Code"] == "ThrottlingException":
            raise  # @aws_throttle handles the backoff/retry
        self.log.error(f"Error adding metadata to {aws_arn}: {type(e).__name__}: {e}")
        return
    except Exception as e:
        self.log.error(f"Error adding metadata to {aws_arn}: {type(e).__name__}: {e}")
        return

    # Poke the modified registry so caches know this artifact changed
    from workbench.cached.cached_meta import CachedMeta

    CachedMeta().update_modified_timestamp(self)

workbench_meta()

Get the Workbench specific metadata for this Artifact

Returns:

Type Description
Union[dict, None]

Union[dict, None]: Dictionary of Workbench metadata for this Artifact
This functionality will work for FeatureSets, Models, and Endpoints

but not for DataSources and Graphs, those classes need to override this method.

Source code in src/workbench/core/artifacts/aws_artifact.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
def workbench_meta(self) -> Union[dict, None]:
    """Get the Workbench specific metadata for this Artifact

    Returns:
        Union[dict, None]: Dictionary of Workbench metadata for this Artifact

    Note: This functionality will work for FeatureSets, Models, and Endpoints
          but not for DataSources and Graphs, those classes need to override this method.
    """
    return self.meta.get_aws_tags(self.arn())