Skip to content

Artifact Stores

ZenML's artifact-store stores artifacts in a file system.

In ZenML, the inputs and outputs which go through any step is treated as an artifact and as its name suggests, an ArtifactStore is a place where these artifacts get stored.

Out of the box, ZenML comes with the BaseArtifactStore and LocalArtifactStore implementations. While the BaseArtifactStore establishes an interface for people who want to extend it to their needs, the LocalArtifactStore is a simple implementation for a local setup.

Moreover, additional artifact stores can be found in specific integrations modules, such as the GCPArtifactStore in the gcp integration and the AzureArtifactStore in the azure integration.

BaseArtifactStore

Bases: StackComponent

Base class for all ZenML artifact stores.

Source code in src/zenml/artifact_stores/base_artifact_store.py
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
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
class BaseArtifactStore(StackComponent):
    """Base class for all ZenML artifact stores."""

    @property
    def config(self) -> BaseArtifactStoreConfig:
        """Returns the `BaseArtifactStoreConfig` config.

        Returns:
            The configuration.
        """
        return cast(BaseArtifactStoreConfig, self._config)

    @property
    def path(self) -> str:
        """The path to the artifact store.

        Returns:
            The path.
        """
        return self.config.path

    @property
    def custom_cache_key(self) -> Optional[bytes]:
        """Custom cache key.

        Any artifact store can override this property in case they need
        additional control over the caching behavior.

        Returns:
            Custom cache key.
        """
        return None

    # --- User interface ---
    @abstractmethod
    def open(self, path: PathType, mode: str = "r") -> Any:
        """Open a file at the given path.

        Args:
            path: The path of the file to open.
            mode: The mode to open the file.

        Returns:
            The file object.
        """

    @abstractmethod
    def copyfile(
        self, src: PathType, dst: PathType, overwrite: bool = False
    ) -> None:
        """Copy a file from the source to the destination.

        Args:
            src: The source path.
            dst: The destination path.
            overwrite: Whether to overwrite the destination file if it exists.
        """

    @abstractmethod
    def exists(self, path: PathType) -> bool:
        """Checks if a path exists.

        Args:
            path: The path to check.

        Returns:
            `True` if the path exists.
        """

    @abstractmethod
    def glob(self, pattern: PathType) -> List[PathType]:
        """Gets the paths that match a glob pattern.

        Args:
            pattern: The glob pattern.

        Returns:
            The list of paths that match the pattern.
        """

    @abstractmethod
    def isdir(self, path: PathType) -> bool:
        """Returns whether the given path points to a directory.

        Args:
            path: The path to check.

        Returns:
            `True` if the path points to a directory.
        """

    @abstractmethod
    def listdir(self, path: PathType) -> List[PathType]:
        """Returns a list of files under a given directory in the filesystem.

        Args:
            path: The path to list.

        Returns:
            The list of files under the given path.
        """

    @abstractmethod
    def makedirs(self, path: PathType) -> None:
        """Make a directory at the given path, recursively creating parents.

        Args:
            path: The path to create.
        """

    @abstractmethod
    def mkdir(self, path: PathType) -> None:
        """Make a directory at the given path; parent directory must exist.

        Args:
            path: The path to create.
        """

    @abstractmethod
    def remove(self, path: PathType) -> None:
        """Remove the file at the given path. Dangerous operation.

        Args:
            path: The path to remove.
        """

    @abstractmethod
    def rename(
        self, src: PathType, dst: PathType, overwrite: bool = False
    ) -> None:
        """Rename source file to destination file.

        Args:
            src: The source path.
            dst: The destination path.
            overwrite: Whether to overwrite the destination file if it exists.
        """

    @abstractmethod
    def rmtree(self, path: PathType) -> None:
        """Deletes dir recursively. Dangerous operation.

        Args:
            path: The path to delete.
        """

    @abstractmethod
    def stat(self, path: PathType) -> Any:
        """Return the stat descriptor for a given file path.

        Args:
            path: The path to check.

        Returns:
            The stat descriptor.
        """

    @abstractmethod
    def size(self, path: PathType) -> Optional[int]:
        """Get the size of a file in bytes.

        Args:
            path: The path to the file.

        Returns:
            The size of the file in bytes or `None` if the artifact store
            does not implement the `size` method.
        """
        logger.warning(
            "Cannot get size of file '%s' since the artifact store %s does not "
            "implement the `size` method.",
            path,
            self.__class__.__name__,
        )
        return None

    @abstractmethod
    def walk(
        self,
        top: PathType,
        topdown: bool = True,
        onerror: Optional[Callable[..., None]] = None,
    ) -> Iterable[Tuple[PathType, List[PathType], List[PathType]]]:
        """Return an iterator that walks the contents of the given directory.

        Args:
            top: The path to walk.
            topdown: Whether to walk the top-down or bottom-up.
            onerror: The error handler.

        Returns:
            The iterator that walks the contents of the given directory.
        """

    # --- Internal interface ---
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initiate the Pydantic object and register the corresponding filesystem.

        Args:
            *args: The positional arguments to pass to the Pydantic object.
            **kwargs: The keyword arguments to pass to the Pydantic object.
        """
        super(BaseArtifactStore, self).__init__(*args, **kwargs)

        # If running in a ZenML server environment, we don't register
        # the filesystems. We always use the artifact stores directly.
        if ENV_ZENML_SERVER not in os.environ:
            self._register()

    def _register(self) -> None:
        """Create and register a filesystem within the filesystem registry."""
        from zenml.io.filesystem import BaseFilesystem
        from zenml.io.filesystem_registry import default_filesystem_registry
        from zenml.io.local_filesystem import LocalFilesystem

        overloads: Dict[str, Any] = {
            "SUPPORTED_SCHEMES": self.config.SUPPORTED_SCHEMES,
        }
        for abc_method in inspect.getmembers(BaseArtifactStore):
            if getattr(abc_method[1], "__isabstractmethod__", False):
                sanitized_method = _sanitize_paths(
                    getattr(self, abc_method[0]), self.path
                )
                # prepare overloads for filesystem methods
                overloads[abc_method[0]] = staticmethod(sanitized_method)

                # decorate artifact store methods
                setattr(
                    self,
                    abc_method[0],
                    sanitized_method,
                )

        # Local filesystem is always registered, no point in doing it again.
        if isinstance(self, LocalFilesystem):
            return

        filesystem_class = type(
            self.__class__.__name__, (BaseFilesystem,), overloads
        )

        default_filesystem_registry.register(filesystem_class)

    def _remove_previous_file_versions(self, path: PathType) -> None:
        """Remove all file versions but the latest in the given path.

        Method is useful for logs stored in versioned file systems
        like AWS S3.

        Args:
            path: The path to the file.
        """
        return

config property

Returns the BaseArtifactStoreConfig config.

Returns:

Type Description
BaseArtifactStoreConfig

The configuration.

custom_cache_key property

Custom cache key.

Any artifact store can override this property in case they need additional control over the caching behavior.

Returns:

Type Description
Optional[bytes]

Custom cache key.

path property

The path to the artifact store.

Returns:

Type Description
str

The path.

__init__(*args, **kwargs)

Initiate the Pydantic object and register the corresponding filesystem.

Parameters:

Name Type Description Default
*args Any

The positional arguments to pass to the Pydantic object.

()
**kwargs Any

The keyword arguments to pass to the Pydantic object.

{}
Source code in src/zenml/artifact_stores/base_artifact_store.py
430
431
432
433
434
435
436
437
438
439
440
441
442
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initiate the Pydantic object and register the corresponding filesystem.

    Args:
        *args: The positional arguments to pass to the Pydantic object.
        **kwargs: The keyword arguments to pass to the Pydantic object.
    """
    super(BaseArtifactStore, self).__init__(*args, **kwargs)

    # If running in a ZenML server environment, we don't register
    # the filesystems. We always use the artifact stores directly.
    if ENV_ZENML_SERVER not in os.environ:
        self._register()

copyfile(src, dst, overwrite=False) abstractmethod

Copy a file from the source to the destination.

Parameters:

Name Type Description Default
src PathType

The source path.

required
dst PathType

The destination path.

required
overwrite bool

Whether to overwrite the destination file if it exists.

False
Source code in src/zenml/artifact_stores/base_artifact_store.py
281
282
283
284
285
286
287
288
289
290
291
@abstractmethod
def copyfile(
    self, src: PathType, dst: PathType, overwrite: bool = False
) -> None:
    """Copy a file from the source to the destination.

    Args:
        src: The source path.
        dst: The destination path.
        overwrite: Whether to overwrite the destination file if it exists.
    """

exists(path) abstractmethod

Checks if a path exists.

Parameters:

Name Type Description Default
path PathType

The path to check.

required

Returns:

Type Description
bool

True if the path exists.

Source code in src/zenml/artifact_stores/base_artifact_store.py
293
294
295
296
297
298
299
300
301
302
@abstractmethod
def exists(self, path: PathType) -> bool:
    """Checks if a path exists.

    Args:
        path: The path to check.

    Returns:
        `True` if the path exists.
    """

glob(pattern) abstractmethod

Gets the paths that match a glob pattern.

Parameters:

Name Type Description Default
pattern PathType

The glob pattern.

required

Returns:

Type Description
List[PathType]

The list of paths that match the pattern.

Source code in src/zenml/artifact_stores/base_artifact_store.py
304
305
306
307
308
309
310
311
312
313
@abstractmethod
def glob(self, pattern: PathType) -> List[PathType]:
    """Gets the paths that match a glob pattern.

    Args:
        pattern: The glob pattern.

    Returns:
        The list of paths that match the pattern.
    """

isdir(path) abstractmethod

Returns whether the given path points to a directory.

Parameters:

Name Type Description Default
path PathType

The path to check.

required

Returns:

Type Description
bool

True if the path points to a directory.

Source code in src/zenml/artifact_stores/base_artifact_store.py
315
316
317
318
319
320
321
322
323
324
@abstractmethod
def isdir(self, path: PathType) -> bool:
    """Returns whether the given path points to a directory.

    Args:
        path: The path to check.

    Returns:
        `True` if the path points to a directory.
    """

listdir(path) abstractmethod

Returns a list of files under a given directory in the filesystem.

Parameters:

Name Type Description Default
path PathType

The path to list.

required

Returns:

Type Description
List[PathType]

The list of files under the given path.

Source code in src/zenml/artifact_stores/base_artifact_store.py
326
327
328
329
330
331
332
333
334
335
@abstractmethod
def listdir(self, path: PathType) -> List[PathType]:
    """Returns a list of files under a given directory in the filesystem.

    Args:
        path: The path to list.

    Returns:
        The list of files under the given path.
    """

makedirs(path) abstractmethod

Make a directory at the given path, recursively creating parents.

Parameters:

Name Type Description Default
path PathType

The path to create.

required
Source code in src/zenml/artifact_stores/base_artifact_store.py
337
338
339
340
341
342
343
@abstractmethod
def makedirs(self, path: PathType) -> None:
    """Make a directory at the given path, recursively creating parents.

    Args:
        path: The path to create.
    """

mkdir(path) abstractmethod

Make a directory at the given path; parent directory must exist.

Parameters:

Name Type Description Default
path PathType

The path to create.

required
Source code in src/zenml/artifact_stores/base_artifact_store.py
345
346
347
348
349
350
351
@abstractmethod
def mkdir(self, path: PathType) -> None:
    """Make a directory at the given path; parent directory must exist.

    Args:
        path: The path to create.
    """

open(path, mode='r') abstractmethod

Open a file at the given path.

Parameters:

Name Type Description Default
path PathType

The path of the file to open.

required
mode str

The mode to open the file.

'r'

Returns:

Type Description
Any

The file object.

Source code in src/zenml/artifact_stores/base_artifact_store.py
269
270
271
272
273
274
275
276
277
278
279
@abstractmethod
def open(self, path: PathType, mode: str = "r") -> Any:
    """Open a file at the given path.

    Args:
        path: The path of the file to open.
        mode: The mode to open the file.

    Returns:
        The file object.
    """

remove(path) abstractmethod

Remove the file at the given path. Dangerous operation.

Parameters:

Name Type Description Default
path PathType

The path to remove.

required
Source code in src/zenml/artifact_stores/base_artifact_store.py
353
354
355
356
357
358
359
@abstractmethod
def remove(self, path: PathType) -> None:
    """Remove the file at the given path. Dangerous operation.

    Args:
        path: The path to remove.
    """

rename(src, dst, overwrite=False) abstractmethod

Rename source file to destination file.

Parameters:

Name Type Description Default
src PathType

The source path.

required
dst PathType

The destination path.

required
overwrite bool

Whether to overwrite the destination file if it exists.

False
Source code in src/zenml/artifact_stores/base_artifact_store.py
361
362
363
364
365
366
367
368
369
370
371
@abstractmethod
def rename(
    self, src: PathType, dst: PathType, overwrite: bool = False
) -> None:
    """Rename source file to destination file.

    Args:
        src: The source path.
        dst: The destination path.
        overwrite: Whether to overwrite the destination file if it exists.
    """

rmtree(path) abstractmethod

Deletes dir recursively. Dangerous operation.

Parameters:

Name Type Description Default
path PathType

The path to delete.

required
Source code in src/zenml/artifact_stores/base_artifact_store.py
373
374
375
376
377
378
379
@abstractmethod
def rmtree(self, path: PathType) -> None:
    """Deletes dir recursively. Dangerous operation.

    Args:
        path: The path to delete.
    """

size(path) abstractmethod

Get the size of a file in bytes.

Parameters:

Name Type Description Default
path PathType

The path to the file.

required

Returns:

Type Description
Optional[int]

The size of the file in bytes or None if the artifact store

Optional[int]

does not implement the size method.

Source code in src/zenml/artifact_stores/base_artifact_store.py
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
@abstractmethod
def size(self, path: PathType) -> Optional[int]:
    """Get the size of a file in bytes.

    Args:
        path: The path to the file.

    Returns:
        The size of the file in bytes or `None` if the artifact store
        does not implement the `size` method.
    """
    logger.warning(
        "Cannot get size of file '%s' since the artifact store %s does not "
        "implement the `size` method.",
        path,
        self.__class__.__name__,
    )
    return None

stat(path) abstractmethod

Return the stat descriptor for a given file path.

Parameters:

Name Type Description Default
path PathType

The path to check.

required

Returns:

Type Description
Any

The stat descriptor.

Source code in src/zenml/artifact_stores/base_artifact_store.py
381
382
383
384
385
386
387
388
389
390
@abstractmethod
def stat(self, path: PathType) -> Any:
    """Return the stat descriptor for a given file path.

    Args:
        path: The path to check.

    Returns:
        The stat descriptor.
    """

walk(top, topdown=True, onerror=None) abstractmethod

Return an iterator that walks the contents of the given directory.

Parameters:

Name Type Description Default
top PathType

The path to walk.

required
topdown bool

Whether to walk the top-down or bottom-up.

True
onerror Optional[Callable[..., None]]

The error handler.

None

Returns:

Type Description
Iterable[Tuple[PathType, List[PathType], List[PathType]]]

The iterator that walks the contents of the given directory.

Source code in src/zenml/artifact_stores/base_artifact_store.py
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
@abstractmethod
def walk(
    self,
    top: PathType,
    topdown: bool = True,
    onerror: Optional[Callable[..., None]] = None,
) -> Iterable[Tuple[PathType, List[PathType], List[PathType]]]:
    """Return an iterator that walks the contents of the given directory.

    Args:
        top: The path to walk.
        topdown: Whether to walk the top-down or bottom-up.
        onerror: The error handler.

    Returns:
        The iterator that walks the contents of the given directory.
    """

BaseArtifactStoreConfig

Bases: StackComponentConfig

Config class for BaseArtifactStore.

Source code in src/zenml/artifact_stores/base_artifact_store.py
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
class BaseArtifactStoreConfig(StackComponentConfig):
    """Config class for `BaseArtifactStore`."""

    path: str

    SUPPORTED_SCHEMES: ClassVar[Set[str]]
    IS_IMMUTABLE_FILESYSTEM: ClassVar[bool] = False

    @model_validator(mode="before")
    @classmethod
    @before_validator_handler
    def _ensure_artifact_store(cls, data: Dict[str, Any]) -> Dict[str, Any]:
        """Validator function for the Artifact Stores.

        Checks whether supported schemes are defined and the given path is
        supported.

        Args:
            data: the input data to construct the artifact store.

        Returns:
            The validated values.

        Raises:
            ArtifactStoreInterfaceError: If the scheme is not supported.
        """
        try:
            getattr(cls, "SUPPORTED_SCHEMES")
        except AttributeError:
            raise ArtifactStoreInterfaceError(
                textwrap.dedent(
                    """
                    When you are working with any classes which subclass from
                    zenml.artifact_store.BaseArtifactStore please make sure
                    that your class has a ClassVar named `SUPPORTED_SCHEMES`
                    which should hold a set of supported file schemes such
                    as {"s3://"} or {"gcs://"}.

                    Example:

                    class MyArtifactStoreConfig(BaseArtifactStoreConfig):
                        ...
                        # Class Variables
                        SUPPORTED_SCHEMES: ClassVar[Set[str]] = {"s3://"}
                        ...
                    """
                )
            )

        if "path" in data:
            data["path"] = data["path"].strip("'\"`")
            if not any(
                data["path"].startswith(i) for i in cls.SUPPORTED_SCHEMES
            ):
                raise ArtifactStoreInterfaceError(
                    f"The path: '{data['path']}' you defined for your "
                    f"artifact store is not supported by the implementation of "
                    f"{cls.schema()['title']}, because it does not start with "
                    f"one of its supported schemes: {cls.SUPPORTED_SCHEMES}."
                )

        return data

BaseArtifactStoreFlavor

Bases: Flavor

Base class for artifact store flavors.

Source code in src/zenml/artifact_stores/base_artifact_store.py
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
class BaseArtifactStoreFlavor(Flavor):
    """Base class for artifact store flavors."""

    @property
    def type(self) -> StackComponentType:
        """Returns the flavor type.

        Returns:
            The flavor type.
        """
        return StackComponentType.ARTIFACT_STORE

    @property
    def config_class(self) -> Type[StackComponentConfig]:
        """Config class for this flavor.

        Returns:
            The config class.
        """
        return BaseArtifactStoreConfig

    @property
    @abstractmethod
    def implementation_class(self) -> Type["BaseArtifactStore"]:
        """Implementation class.

        Returns:
            The implementation class.
        """

config_class property

Config class for this flavor.

Returns:

Type Description
Type[StackComponentConfig]

The config class.

implementation_class abstractmethod property

Implementation class.

Returns:

Type Description
Type[BaseArtifactStore]

The implementation class.

type property

Returns the flavor type.

Returns:

Type Description
StackComponentType

The flavor type.

LocalArtifactStore

Bases: LocalFilesystem, BaseArtifactStore

Artifact Store for local artifacts.

All methods are inherited from the default LocalFilesystem.

Source code in src/zenml/artifact_stores/local_artifact_store.py
 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
class LocalArtifactStore(LocalFilesystem, BaseArtifactStore):
    """Artifact Store for local artifacts.

    All methods are inherited from the default `LocalFilesystem`.
    """

    _path: Optional[str] = None

    @staticmethod
    def get_default_local_path(id_: "UUID") -> str:
        """Returns the default local path for a local artifact store.

        Args:
            id_: The id of the local artifact store.

        Returns:
            str: The default local path.
        """
        return os.path.join(
            GlobalConfiguration().local_stores_path,
            str(id_),
        )

    @property
    def path(self) -> str:
        """Returns the path to the local artifact store.

        If the user has not defined a path in the config, this will create a
        sub-folder in the global config directory.

        Returns:
            The path to the local artifact store.
        """
        if self._path:
            return self._path

        if self.config.path:
            self._path = self.config.path
        else:
            self._path = self.get_default_local_path(self.id)
        io_utils.create_dir_recursive_if_not_exists(self._path)
        return self._path

    @property
    def local_path(self) -> Optional[str]:
        """Returns the local path of the artifact store.

        Returns:
            The local path of the artifact store.
        """
        return self.path

    @property
    def custom_cache_key(self) -> Optional[bytes]:
        """Custom cache key.

        The client ID is returned here to invalidate caching when using the same
        local artifact store on multiple client machines.

        Returns:
            Custom cache key.
        """
        return GlobalConfiguration().user_id.bytes

custom_cache_key property

Custom cache key.

The client ID is returned here to invalidate caching when using the same local artifact store on multiple client machines.

Returns:

Type Description
Optional[bytes]

Custom cache key.

local_path property

Returns the local path of the artifact store.

Returns:

Type Description
Optional[str]

The local path of the artifact store.

path property

Returns the path to the local artifact store.

If the user has not defined a path in the config, this will create a sub-folder in the global config directory.

Returns:

Type Description
str

The path to the local artifact store.

get_default_local_path(id_) staticmethod

Returns the default local path for a local artifact store.

Parameters:

Name Type Description Default
id_ UUID

The id of the local artifact store.

required

Returns:

Name Type Description
str str

The default local path.

Source code in src/zenml/artifact_stores/local_artifact_store.py
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@staticmethod
def get_default_local_path(id_: "UUID") -> str:
    """Returns the default local path for a local artifact store.

    Args:
        id_: The id of the local artifact store.

    Returns:
        str: The default local path.
    """
    return os.path.join(
        GlobalConfiguration().local_stores_path,
        str(id_),
    )

LocalArtifactStoreConfig

Bases: BaseArtifactStoreConfig

Config class for the local artifact store.

Attributes:

Name Type Description
path str

The path to the local artifact store.

Source code in src/zenml/artifact_stores/local_artifact_store.py
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
class LocalArtifactStoreConfig(BaseArtifactStoreConfig):
    """Config class for the local artifact store.

    Attributes:
        path: The path to the local artifact store.
    """

    SUPPORTED_SCHEMES: ClassVar[Set[str]] = {""}

    path: str = ""

    @field_validator("path")
    @classmethod
    def ensure_path_local(cls, path: str) -> str:
        """Pydantic validator which ensures that the given path is a local path.

        Args:
            path: The path to validate.

        Returns:
            str: The validated (local) path.

        Raises:
            ArtifactStoreInterfaceError: If the given path is not a local path.
        """
        remote_prefixes = ["gs://", "hdfs://", "s3://", "az://", "abfs://"]
        if any(path.startswith(prefix) for prefix in remote_prefixes):
            raise ArtifactStoreInterfaceError(
                f"The path '{path}' you defined for your local artifact store "
                f"starts with a remote prefix."
            )
        return path

    @property
    def is_local(self) -> bool:
        """Checks if this stack component is running locally.

        Returns:
            True if this config is for a local component, False otherwise.
        """
        return True

is_local property

Checks if this stack component is running locally.

Returns:

Type Description
bool

True if this config is for a local component, False otherwise.

ensure_path_local(path) classmethod

Pydantic validator which ensures that the given path is a local path.

Parameters:

Name Type Description Default
path str

The path to validate.

required

Returns:

Name Type Description
str str

The validated (local) path.

Raises:

Type Description
ArtifactStoreInterfaceError

If the given path is not a local path.

Source code in src/zenml/artifact_stores/local_artifact_store.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@field_validator("path")
@classmethod
def ensure_path_local(cls, path: str) -> str:
    """Pydantic validator which ensures that the given path is a local path.

    Args:
        path: The path to validate.

    Returns:
        str: The validated (local) path.

    Raises:
        ArtifactStoreInterfaceError: If the given path is not a local path.
    """
    remote_prefixes = ["gs://", "hdfs://", "s3://", "az://", "abfs://"]
    if any(path.startswith(prefix) for prefix in remote_prefixes):
        raise ArtifactStoreInterfaceError(
            f"The path '{path}' you defined for your local artifact store "
            f"starts with a remote prefix."
        )
    return path

LocalArtifactStoreFlavor

Bases: BaseArtifactStoreFlavor

Class for the LocalArtifactStoreFlavor.

Source code in src/zenml/artifact_stores/local_artifact_store.py
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
class LocalArtifactStoreFlavor(BaseArtifactStoreFlavor):
    """Class for the `LocalArtifactStoreFlavor`."""

    @property
    def name(self) -> str:
        """Returns the name of the artifact store flavor.

        Returns:
            str: The name of the artifact store flavor.
        """
        return "local"

    @property
    def docs_url(self) -> Optional[str]:
        """A url to point at docs explaining this flavor.

        Returns:
            A flavor docs url.
        """
        return self.generate_default_docs_url()

    @property
    def sdk_docs_url(self) -> Optional[str]:
        """A url to point at docs explaining this flavor.

        Returns:
            A flavor docs url.
        """
        return self.generate_default_sdk_docs_url()

    @property
    def logo_url(self) -> str:
        """A url to represent the flavor in the dashboard.

        Returns:
            The flavor logo.
        """
        return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/artifact_store/local.svg"

    @property
    def config_class(self) -> Type[LocalArtifactStoreConfig]:
        """Config class for this flavor.

        Returns:
            The config class.
        """
        return LocalArtifactStoreConfig

    @property
    def implementation_class(self) -> Type[LocalArtifactStore]:
        """Implementation class.

        Returns:
            The implementation class.
        """
        return LocalArtifactStore

config_class property

Config class for this flavor.

Returns:

Type Description
Type[LocalArtifactStoreConfig]

The config class.

docs_url property

A url to point at docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor docs url.

implementation_class property

Implementation class.

Returns:

Type Description
Type[LocalArtifactStore]

The implementation class.

logo_url property

A url to represent the flavor in the dashboard.

Returns:

Type Description
str

The flavor logo.

name property

Returns the name of the artifact store flavor.

Returns:

Name Type Description
str str

The name of the artifact store flavor.

sdk_docs_url property

A url to point at docs explaining this flavor.

Returns:

Type Description
Optional[str]

A flavor docs url.