Skip to content

Artifact

The hera.workflows.artifact module provides the base Artifact class, along with the various types of artifacts as subclasses.

Tip

Read the Hera walk-through for Artifacts.

Artifact

Base artifact representation.

Source code in src/hera/workflows/artifact.py 
 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
@dataclass(kw_only=True)
class Artifact:
    """Base artifact representation."""

    name: Optional[str] = None
    """name of the artifact"""

    archive: Optional[Union[_ModelArchiveStrategy, ArchiveStrategy]] = None
    """artifact archiving configuration"""

    archive_logs: Optional[bool] = None
    """whether to log the archive object"""

    artifact_gc: Optional[ArtifactGC] = None
    """artifact garbage collection configuration"""

    deleted: Optional[bool] = None
    """whether the artifact is deleted"""

    from_: Optional[str] = None
    """configures the artifact task/step origin"""

    from_expression: Optional[str] = None
    """an expression that dictates where to obtain the artifact from"""

    global_name: Optional[str] = None
    """global workflow artifact name"""

    mode: Optional[int] = None
    """mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts."""

    path: Optional[str] = None
    """path where the artifact should be placed/loaded from"""

    recurse_mode: Optional[bool] = None
    """recursion mode when applying the permissions of the artifact if it is an artifact folder"""

    sub_path: Optional[str] = None
    """allows the specification of an artifact from a subpath within the main source."""

    loader: Optional[ArtifactLoader] = None
    """used for input Artifact annotations for determining how to load the data.

    Note: A loader value of 'None' must be used with an underlying type of 'str' or Path-like class."""

    loadb: Optional[Callable[[bytes], Any]] = None
    """used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters"""

    dumpb: Optional[Callable[[Any], bytes]] = None
    """used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters"""

    loads: Optional[Callable[[str], Any]] = None
    """used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters"""

    dumps: Optional[Callable[[Any], str]] = None
    """used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters"""

    optional: Optional[bool] = None
    """whether the Artifact is optional. For an input Artifact, this means it may possibly not
    exist at the specified path during the template's runtime. For an output Artifact, it may
    possibly not be generated during the step/task and available as an output to subsequent steps/tasks."""

    output: bool = False
    """used to specify artifact as an output in function signature annotations"""

    def _check_name(self):
        if not self.name:
            raise ValueError("name cannot be `None` or empty when used")

    def _get_default_inputs_path(self) -> str:
        return _DEFAULT_ARTIFACT_INPUT_DIRECTORY + f"{self.name}"

    def _build_archive(self) -> Optional[_ModelArchiveStrategy]:
        if self.archive is None:
            return None

        if isinstance(self.archive, _ModelArchiveStrategy):
            return self.archive
        return cast(ArchiveStrategy, self.archive)._build_archive_strategy()

    def _build_artifact(self) -> _ModelArtifact:
        self._check_name()
        assert self.name
        return _ModelArtifact(
            name=self.name,
            archive=self._build_archive(),
            archive_logs=self.archive_logs,
            artifact_gc=self.artifact_gc,
            deleted=self.deleted,
            from_=self.from_,
            from_expression=self.from_expression,
            global_name=self.global_name,
            mode=self.mode,
            optional=self.optional,
            path=self.path,
            recurse_mode=self.recurse_mode,
            sub_path=self.sub_path,
        )

    def _build_artifact_paths(self) -> _ModelArtifactPaths:
        self._check_name()
        artifact = self._build_artifact()
        return _ModelArtifactPaths(**artifact.model_dump())

    def with_name(self, name: str) -> Artifact:
        """Returns a copy of the current artifact, renamed using the specified `name`."""
        artifact = deepcopy(self)
        artifact.name = name
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return [
            "mode",
            "name",
            "optional",
            "path",
            "recurseMode",
            "subPath",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

ArtifactLoader

Enum for artifact loader options.

Source code in src/hera/workflows/artifact.py 
48
49
50
51
52
53
54
55
class ArtifactLoader(Enum):
    """Enum for artifact loader options."""

    json = "json"
    """Deserialize the JSON-string Artifact file to a Python object (the target variable can be any JSON-compatible type)."""

    file = "file"
    """Read the contents of the Artifact file directly as a string (the target variable must be a `str` type)."""

file 

file = 'file'

Read the contents of the Artifact file directly as a string (the target variable must be a str type).

json 

json = 'json'

Deserialize the JSON-string Artifact file to a Python object (the target variable can be any JSON-compatible type).

ArtifactoryArtifact

An artifact sourced from Artifactory.

Source code in src/hera/workflows/artifact.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@dataclass(kw_only=True)
class ArtifactoryArtifact(Artifact):
    """An artifact sourced from Artifactory."""

    password_secret: Optional[SecretKeySelector] = None
    url: str
    username_secret: Optional[SecretKeySelector] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.artifactory = _ModelArtifactoryArtifact(
            url=self.url, password_secret=self.password_secret, username_secret=self.username_secret
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + ["url", "password_secret", "username_secret"]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

AzureArtifact

An artifact sourced from Microsoft Azure.

Source code in src/hera/workflows/artifact.py 
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
@dataclass(kw_only=True)
class AzureArtifact(Artifact):
    """An artifact sourced from Microsoft Azure."""

    account_key_secret: Optional[SecretKeySelector] = None
    blob: str
    container: str
    endpoint: str
    use_sdk_creds: Optional[bool] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.azure = _ModelAzureArtifact(
            account_key_secret=self.account_key_secret,
            blob=self.blob,
            container=self.container,
            endpoint=self.endpoint,
            use_sdk_creds=self.use_sdk_creds,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + [
            "endpoint",
            "container",
            "blob",
            "account_key_secret",
            "use_sdk_creds",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

GCSArtifact

An artifact sourced from Google Cloud Storage.

Source code in src/hera/workflows/artifact.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
@dataclass(kw_only=True)
class GCSArtifact(Artifact):
    """An artifact sourced from Google Cloud Storage."""

    bucket: Optional[str] = None
    key: str
    service_account_key_secret: Optional[SecretKeySelector] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.gcs = _ModelGCSArtifact(
            bucket=self.bucket,
            key=self.key,
            service_account_key_secret=self.service_account_key_secret,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + ["bucket", "key", "service_account_key_secret"]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

GitArtifact

An artifact sourced from GitHub.

Source code in src/hera/workflows/artifact.py 
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
@dataclass(kw_only=True)
class GitArtifact(Artifact):
    """An artifact sourced from GitHub."""

    branch: Optional[str] = None
    depth: Optional[int] = None
    disable_submodules: Optional[bool] = None
    fetch: Optional[List[str]] = None
    insecure_ignore_host_key: Optional[bool] = None
    insecure_skip_tls: Optional[bool] = None
    password_secret: Optional[SecretKeySelector] = None
    repo: str
    revision: Optional[str] = None
    single_branch: Optional[bool] = None
    ssh_private_key_secret: Optional[SecretKeySelector] = None
    username_secret: Optional[SecretKeySelector] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.git = _ModelGitArtifact(
            branch=self.branch,
            depth=self.depth,
            disable_submodules=self.disable_submodules,
            fetch=self.fetch,
            insecure_ignore_host_key=self.insecure_ignore_host_key,
            password_secret=self.password_secret,
            repo=self.repo,
            revision=self.revision,
            single_branch=self.single_branch,
            ssh_private_key_secret=self.ssh_private_key_secret,
            username_secret=self.username_secret,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + [
            "branch",
            "depth",
            "disable_submodules",
            "fetch",
            "insecure_ignore_host_key",
            "password_secret",
            "repo",
            "revision",
            "single_branch",
            "ssh_private_key_secret",
            "username_secret",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

HDFSArtifact

A Hadoop File System artifact.

Note that we redefine the HDFS path to hdfs_path to differentiate between the parent Artifact class path and the child class path.

Source code in src/hera/workflows/artifact.py 
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
@dataclass(kw_only=True)
class HDFSArtifact(Artifact):
    """A Hadoop File System artifact.

    Note that we redefine the HDFS `path` to `hdfs_path` to differentiate between the
    parent Artifact class `path` and the child class `path`.
    """

    hdfs_path: str
    addresses: Optional[List[str]] = None
    force: Optional[bool] = None
    hdfs_user: Optional[str]
    krb_c_cache_secret: Optional[SecretKeySelector] = None
    krb_config_config_map: Optional[ConfigMapKeySelector] = None
    krb_keytab_secret: Optional[SecretKeySelector] = None
    krb_realm: Optional[str] = None
    krb_service_principal_name: Optional[str] = None
    krb_username: Optional[str] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.hdfs = _ModelHDFSArtifact(
            addresses=self.addresses,
            force=self.force,
            hdfs_user=self.hdfs_user,
            krb_c_cache_secret=self.krb_c_cache_secret,
            krb_config_config_map=self.krb_config_config_map,
            krb_keytab_secret=self.krb_keytab_secret,
            krb_realm=self.krb_realm,
            krb_service_principal_name=self.krb_service_principal_name,
            krb_username=self.krb_username,
            path=self.hdfs_path,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + [
            "addresses",
            "force",
            "hdfs_path",
            "hdfs_user",
            "krb_c_cache_secret",
            "krb_config_config_map",
            "krb_keytab_secret",
            "krb_realm",
            "krb_service_principal_name",
            "krb_username",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

HTTPArtifact

An artifact sourced from an HTTP URL.

Source code in src/hera/workflows/artifact.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
@dataclass(kw_only=True)
class HTTPArtifact(Artifact):
    """An artifact sourced from an HTTP URL."""

    auth: Optional[HTTPAuth] = None
    headers: Optional[List[Header]] = None
    url: str

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.http = _ModelHTTPArtifact(
            auth=self.auth,
            headers=self.headers,
            url=self.url,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + ["auth", "headers", "url"]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

OSSArtifact

An artifact sourced from OSS.

Source code in src/hera/workflows/artifact.py 
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
@dataclass(kw_only=True)
class OSSArtifact(Artifact):
    """An artifact sourced from OSS."""

    access_key_secret: Optional[SecretKeySelector] = None
    bucket: Optional[str] = None
    create_bucket_if_not_present: Optional[bool] = None
    endpoint: Optional[str] = None
    key: str
    lifecycle_rule: Optional[OSSLifecycleRule] = None
    secret_key_secret: Optional[SecretKeySelector] = None
    security_token: Optional[str] = None
    use_sdk_creds: Optional[bool] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.oss = _ModelOSSArtifact(
            access_key_secret=self.access_key_secret,
            bucket=self.bucket,
            create_bucket_if_not_present=self.create_bucket_if_not_present,
            endpoint=self.endpoint,
            key=self.key,
            lifecycle_rule=self.lifecycle_rule,
            secret_key_secret=self.secret_key_secret,
            security_token=self.security_token,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + [
            "access_key_secret",
            "bucket",
            "create_bucket_if_not_present",
            "endpoint",
            "key",
            "lifecycle_rule",
            "secret_key_secret",
            "security_token",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

RawArtifact

A raw bytes artifact representation.

Source code in src/hera/workflows/artifact.py 
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
@dataclass(kw_only=True)
class RawArtifact(Artifact):
    """A raw bytes artifact representation."""

    data: str

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.raw = _ModelRawArtifact(data=self.data)
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + ["data"]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

S3Artifact

An artifact sourced from AWS S3.

Source code in src/hera/workflows/artifact.py 
479
480
481
482
483
484
485
486
487
488
489
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
519
520
521
522
523
524
525
526
527
528
529
@dataclass(kw_only=True)
class S3Artifact(Artifact):
    """An artifact sourced from AWS S3."""

    access_key_secret: Optional[SecretKeySelector] = None
    bucket: Optional[str] = None
    ca_secret: Optional[SecretKeySelector] = None
    create_bucket_if_not_present: Optional[CreateS3BucketOptions] = None
    encryption_options: Optional[S3EncryptionOptions] = None
    endpoint: Optional[str] = None
    insecure: Optional[bool] = None
    key: Optional[str] = None
    region: Optional[str] = None
    role_arn: Optional[str] = None
    secret_key_secret: Optional[SecretKeySelector] = None
    session_token_secret: Optional[SecretKeySelector] = None
    use_sdk_creds: Optional[bool] = None

    def _build_artifact(self) -> _ModelArtifact:
        artifact = super()._build_artifact()
        artifact.s3 = _ModelS3Artifact(
            access_key_secret=self.access_key_secret,
            bucket=self.bucket,
            create_bucket_if_not_present=self.create_bucket_if_not_present,
            encryption_options=self.encryption_options,
            endpoint=self.endpoint,
            insecure=self.insecure,
            key=self.key,
            region=self.region,
            role_arn=self.role_arn,
            secret_key_secret=self.secret_key_secret,
            use_sdk_creds=self.use_sdk_creds,
        )
        return artifact

    @classmethod
    def _get_input_attributes(cls):
        """Return the attributes used for input artifact annotations."""
        return super()._get_input_attributes() + [
            "access_key_secret",
            "bucket",
            "create_bucket_if_not_present",
            "encryption_options",
            "endpoint",
            "insecure",
            "key",
            "region",
            "role_arn",
            "secret_key_secret",
            "use_sdk_creds",
        ]

archive 

archive: Optional[
    Union[ArchiveStrategy, ArchiveStrategy]
] = None

artifact archiving configuration

archive_logs 

archive_logs: Optional[bool] = None

whether to log the archive object

artifact_gc 

artifact_gc: Optional[ArtifactGC] = None

artifact garbage collection configuration

deleted 

deleted: Optional[bool] = None

whether the artifact is deleted

dumpb 

dumpb: Optional[Callable[[Any], bytes]] = None

used to specify a dumper function to serialise the Artifact value as bytes for Annotated Artifact function parameters

dumps 

dumps: Optional[Callable[[Any], str]] = None

used to specify a dumper function to serialise the Artifact value as a string for Annotated Artifact function parameters

from_ 

from_: Optional[str] = None

configures the artifact task/step origin

from_expression 

from_expression: Optional[str] = None

an expression that dictates where to obtain the artifact from

global_name 

global_name: Optional[str] = None

global workflow artifact name

loadb 

loadb: Optional[Callable[[bytes], Any]] = None

used to specify a loader function to deserialise from bytes for Annotated Artifact function parameters

loader 

loader: Optional[ArtifactLoader] = None

used for input Artifact annotations for determining how to load the data.

Note: A loader value of ‘None’ must be used with an underlying type of ‘str’ or Path-like class.

loads 

loads: Optional[Callable[[str], Any]] = None

used to specify a loader function to deserialise a string representation of an object for Annotated Artifact function parameters

mode 

mode: Optional[int] = None

mode bits to use on the artifact, must be a value between 0 and 0777 set when loading input artifacts.

name 

name: Optional[str] = None

name of the artifact

optional 

optional: Optional[bool] = None

whether the Artifact is optional. For an input Artifact, this means it may possibly not exist at the specified path during the template’s runtime. For an output Artifact, it may possibly not be generated during the step/task and available as an output to subsequent steps/tasks.

output 

output: bool = False

used to specify artifact as an output in function signature annotations

path 

path: Optional[str] = None

path where the artifact should be placed/loaded from

recurse_mode 

recurse_mode: Optional[bool] = None

recursion mode when applying the permissions of the artifact if it is an artifact folder

sub_path 

sub_path: Optional[str] = None

allows the specification of an artifact from a subpath within the main source.

with_name 

with_name(name: str) -> Artifact

Returns a copy of the current artifact, renamed using the specified name.

Source code in src/hera/workflows/artifact.py 
162
163
164
165
166
def with_name(self, name: str) -> Artifact:
    """Returns a copy of the current artifact, renamed using the specified `name`."""
    artifact = deepcopy(self)
    artifact.name = name
    return artifact

Comments