Artifact¶

An Artifact represents a binary data blob stored in IVCAP — images, CSV files, NetCDF datasets, model checkpoints, etc. Artifacts are created by ivcap.upload_artifact() or ivcap.get_artifact().

Quick Reference¶

from ivcap_client.ivcap import IVCAP

ivcap = IVCAP()

# Upload a file
artifact = ivcap.upload_artifact(
    name="my-data",
    file_path="/path/to/file.csv",
)
print(artifact.id, artifact.mime_type)

# Download
with artifact.as_local_file() as path:
    print(f"Downloaded to: {path}")

# Stream to disk
with open("/tmp/output.csv", "wb") as f:
    for chunk in artifact.as_stream():
        f.write(chunk)

Class Documentation¶

`Artifact` `dataclass` ¶

Represents an artifact stored in an IVCAP deployment.

An artifact is any binary or structured data blob produced or consumed by a job — an image, a CSV file, a NetCDF dataset, a trained model checkpoint, etc.

Each artifact has two complementary parts:

Blob — the raw bytes, stored in object storage (GCS/S3-compatible).
Aspects — typed metadata records in the Datafabric describing the artifact's MIME type, size, provenance, and any domain annotations.

Key properties:

id / urn — canonical urn:ivcap:artifact:<uuid> identifier
name — human-readable name
mime_type — MIME content type (e.g. "image/jpeg")
size — size in bytes
status — current :class:~ivcap_client.models.ArtifactStatusRTStatus value

Reading artifact content:

:meth:as_local_file — recommended download method; saves to a temp file (auto-deleted on context exit) or to an explicit path.
:meth:open — returns a file-like object with all bytes loaded into memory (convenient for small files).
:meth:as_stream — yields raw bytes chunks for memory-efficient streaming or custom chunk processing.

Example::

artifact = ivcap.get_artifact("urn:ivcap:artifact:<uuid>")

# Download to a temp file (auto-deleted when the 'with' block exits)
with artifact.as_local_file() as path:
    data = path.read_bytes()

# Download to a specific path (file is kept)
path = artifact.as_local_file("/tmp/output.jpg")

# Load entirely into memory
with artifact.open() as f:
    data = f.read()

Source code in ivcap_client/artifact.py

@dataclass
class Artifact:
    """Represents an artifact stored in an IVCAP deployment.

    An artifact is any binary or structured data blob produced or consumed by a job —
    an image, a CSV file, a NetCDF dataset, a trained model checkpoint, etc.

    Each artifact has two complementary parts:

    1. **Blob** — the raw bytes, stored in object storage (GCS/S3-compatible).
    2. **Aspects** — typed metadata records in the Datafabric describing the artifact's
       MIME type, size, provenance, and any domain annotations.

    Key properties:

    * ``id`` / ``urn`` — canonical ``urn:ivcap:artifact:<uuid>`` identifier
    * ``name`` — human-readable name
    * ``mime_type`` — MIME content type (e.g. ``"image/jpeg"``)
    * ``size`` — size in bytes
    * ``status`` — current :class:`~ivcap_client.models.ArtifactStatusRTStatus` value

    Reading artifact content:

    * :meth:`as_local_file` — **recommended** download method; saves to a temp file
      (auto-deleted on context exit) or to an explicit path.
    * :meth:`open` — returns a file-like object with all bytes loaded into memory
      (convenient for small files).
    * :meth:`as_stream` — yields raw ``bytes`` chunks for memory-efficient streaming
      or custom chunk processing.

    Example::

        artifact = ivcap.get_artifact("urn:ivcap:artifact:<uuid>")

        # Download to a temp file (auto-deleted when the 'with' block exits)
        with artifact.as_local_file() as path:
            data = path.read_bytes()

        # Download to a specific path (file is kept)
        path = artifact.as_local_file("/tmp/output.jpg")

        # Load entirely into memory
        with artifact.open() as f:
            data = f.read()
    """

    id: str
    status: ArtifactStatusRTStatus
    name: str | None = None
    size: int | None = None
    mime_type: str | None = None
    created_at: datetime.datetime | None = None
    last_modified_at: datetime.datetime | None = None

    etag: str | None = None

    policy: URN | None = None
    account: URN | None = None

    @classmethod
    def _from_list_item(cls, item: ArtifactListItem, ivcap: IVCAP):
        kwargs = item.to_dict()
        return cls(ivcap, **kwargs)

    def __init__(self, ivcap: IVCAP, **kwargs):
        if not ivcap:
            raise ValueError("missing 'ivcap' argument")
        self._ivcap = ivcap
        self.__update__(**kwargs)

    def __update__(self, **kwargs):
        p = [
            "id",
            "name",
            "size",
            "mime-type",
            "last-modified-at",
            "created-at",
            "policy",
            "etag",
            "account",
        ]
        hp = ["status", "cache_of", "data-href"]
        _set_fields(self, p, hp, kwargs)

        if self._data_href:
            self._data_href = fix_data_ref(self._data_href)
        if not self.id:
            raise ValueError("missing 'id' for Artifact")

    @property
    def urn(self) -> str:
        return self.id

    @property
    def status(self, refresh=True) -> ArtifactStatusRT:
        if refresh or not self._status:
            self.refresh()
        return self._status

    def refresh(self) -> Artifact:
        r = artifact_read.sync_detailed(client=self._ivcap._client, id=self.id)
        if r.status_code >= 300:
            return process_error("place_order", r)
        kwargs = r.parsed.to_dict()
        self.__update__(**kwargs)
        return self

    def open(self) -> io.IOBase:
        """Return a file-like object with the full artifact content loaded into memory.

        The entire artifact blob is fetched in a single HTTP request and wrapped in a
        :class:`ProxyFile` backed by an in-memory :class:`io.BytesIO` buffer.  This
        is convenient for small artifacts where memory is not a concern.

        .. warning::
            The full content is loaded into memory.  For large artifacts (hundreds of
            MB or more) prefer :meth:`as_stream` (chunked iteration) or
            :meth:`as_local_file` (stream-to-disk), both of which avoid holding the
            entire blob in RAM.

        Returns:
            ProxyFile: A readable, seekable, context-manager-compatible file-like
            object.  Call :meth:`~ProxyFile.read` or iterate over lines.

        Example::

            with artifact.open() as f:
                data = f.read()           # bytes
                text = data.decode("utf-8")
        """
        client = self._ivcap._client.get_httpx_client()
        response = client.get(self._data_href)
        response.raise_for_status()
        b = io.BytesIO(response.content)
        return ProxyFile(b)

    def as_stream(self, chunk_size: int = -1) -> Iterator[bytes]:
        """Stream the artifact content as a sequence of raw byte chunks.

        Uses an HTTP streaming GET so that only ``chunk_size`` bytes are buffered
        in memory at a time.  This is the lowest-level download method and is
        suitable when you need to:

        * Implement custom progress reporting.
        * Pipe artifact bytes into a third-party streaming API.
        * Process data incrementally without writing a local file.

        For simply saving the artifact to disk, :meth:`as_local_file` is more
        ergonomic.  For loading everything into memory at once, use :meth:`open`.

        Args:
            chunk_size (int): Number of bytes to read per chunk.  Pass ``-1``
                (the default) to let the SDK choose a chunk size automatically:
                ``max(8 KiB, min(artifact_size / 10, 10 MiB))``.  The artifact
                ``size`` attribute is used when available; if it is unknown the
                fallback is 8 KiB.

        Yields:
            bytes: The next chunk of raw artifact bytes.

        Example::

            # Stream to a file with progress reporting
            total = 0
            with open("/tmp/output.dat", "wb") as f:
                for chunk in artifact.as_stream():
                    f.write(chunk)
                    total += len(chunk)
            print(f"Downloaded {total} bytes")
        """
        chunk_size = _resolve_chunk_size(chunk_size, self.size)
        client = self._ivcap._client.get_httpx_client()
        with client.stream("GET", self._data_href) as response:
            response.raise_for_status()
            for chunk in response.iter_bytes(chunk_size=chunk_size):
                yield chunk

    def as_local_file(self, path: Path | None = None, chunk_size: int = -1) -> Path:
        """Download the artifact to a local file and return the path.

        This is the **recommended** method for saving artifact content to disk.
        It supports two usage patterns depending on whether ``path`` is supplied:

        **Temporary file (path omitted)** — a new temp file is created and a
        :class:`CMPath` is returned.  :class:`CMPath` is a context-manager-aware
        :class:`~pathlib.Path` subclass: when used in a ``with`` statement the
        file is **automatically deleted** when the block exits.  Use this pattern
        when you only need the file transiently::

            with artifact.as_local_file() as path:
                data = path.read_bytes()
            # temp file has been deleted here

        **Explicit path (path provided)** — the content is streamed to the given
        path (parent directories are created automatically).  A plain
        :class:`~pathlib.Path` is returned; the file is **not** deleted
        automatically.  Use this pattern when you want to keep the file::

            path = artifact.as_local_file("/tmp/output.jpg")
            # file remains at /tmp/output.jpg

        Args:
            path: Destination file path.  If ``None`` (default), a temporary file
                is created and wrapped in a self-deleting :class:`CMPath`.
                If a path is provided, the file is written there and a plain
                :class:`~pathlib.Path` is returned.
            chunk_size (int): Number of bytes to read per chunk.  Pass ``-1``
                (the default) to let the SDK choose automatically based on the
                artifact size (see :meth:`as_stream` for the formula).

        Returns:
            :class:`CMPath` when ``path`` is ``None`` (context-manager, auto-delete).
            Plain :class:`~pathlib.Path` when ``path`` is provided (persists).

        Note:
            Both :class:`CMPath` and :class:`~pathlib.Path` are
            :class:`~pathlib.Path` subclasses and support all normal path
            operations (``read_bytes()``, ``open()``, etc.).
        """
        if path is not None:
            path = Path(path)
            path.parent.mkdir(parents=True, exist_ok=True)
            with open(path, "wb") as f:
                for chunk in self.as_stream(chunk_size=chunk_size):
                    f.write(chunk)
            return path
        temp_file = tempfile.NamedTemporaryFile(delete=False)
        with temp_file as f:
            for chunk in self.as_stream(chunk_size=chunk_size):
                f.write(chunk)
        return CMPath(temp_file.name)

    @property
    def metadata(self) -> Iterator[Aspect]:
        return self._ivcap.list_aspects(entity=self.id)

    def add_metadata(
        self,
        aspect: dict[str, any],
        *,
        schema: str | None = None,
        policy: URN | None = None,
    ) -> Artifact:
        """Add a metadata 'aspect' to this artifact. The 'schema' of the aspect, if not defined
        is expected to found in the 'aspect' under the '$schema' key.

        Args:
            aspect (dict): The aspect to be attached
            schema (Optional[str], optional): Schema of the aspect. Defaults to 'aspect["$schema"]'.
            policy: Optional[URN]: Set specific policy controlling access ('urn:ivcap:policy:...').

        Returns:
            self: To enable chaining
        """
        self._ivcap.add_aspect(
            entity=self.id, aspect=aspect, schema=schema, policy=policy
        )
        return self

    def __repr__(self):
        return (
            f"<Artifact id={self.id}, status={self._status if self._status else '???'}>"
        )

`open()` ¶

Return a file-like object with the full artifact content loaded into memory.

The entire artifact blob is fetched in a single HTTP request and wrapped in a :class:ProxyFile backed by an in-memory :class:io.BytesIO buffer. This is convenient for small artifacts where memory is not a concern.

.. warning:: The full content is loaded into memory. For large artifacts (hundreds of MB or more) prefer :meth:as_stream (chunked iteration) or :meth:as_local_file (stream-to-disk), both of which avoid holding the entire blob in RAM.

Returns:

Name	Type	Description
`ProxyFile`	`IOBase`	A readable, seekable, context-manager-compatible file-like object. Call :meth:`~ProxyFile.read` or iterate over lines.

Example::

with artifact.open() as f:
    data = f.read()           # bytes
    text = data.decode("utf-8")

Source code in ivcap_client/artifact.py

def open(self) -> io.IOBase:
    """Return a file-like object with the full artifact content loaded into memory.

    The entire artifact blob is fetched in a single HTTP request and wrapped in a
    :class:`ProxyFile` backed by an in-memory :class:`io.BytesIO` buffer.  This
    is convenient for small artifacts where memory is not a concern.

    .. warning::
        The full content is loaded into memory.  For large artifacts (hundreds of
        MB or more) prefer :meth:`as_stream` (chunked iteration) or
        :meth:`as_local_file` (stream-to-disk), both of which avoid holding the
        entire blob in RAM.

    Returns:
        ProxyFile: A readable, seekable, context-manager-compatible file-like
        object.  Call :meth:`~ProxyFile.read` or iterate over lines.

    Example::

        with artifact.open() as f:
            data = f.read()           # bytes
            text = data.decode("utf-8")
    """
    client = self._ivcap._client.get_httpx_client()
    response = client.get(self._data_href)
    response.raise_for_status()
    b = io.BytesIO(response.content)
    return ProxyFile(b)

`as_stream(chunk_size=-1)` ¶

Stream the artifact content as a sequence of raw byte chunks.

Uses an HTTP streaming GET so that only chunk_size bytes are buffered in memory at a time. This is the lowest-level download method and is suitable when you need to:

Implement custom progress reporting.
Pipe artifact bytes into a third-party streaming API.
Process data incrementally without writing a local file.

For simply saving the artifact to disk, :meth:as_local_file is more ergonomic. For loading everything into memory at once, use :meth:open.

Parameters:

Name	Type	Description	Default
`chunk_size`	`int`	Number of bytes to read per chunk. Pass `-1` (the default) to let the SDK choose a chunk size automatically: `max(8 KiB, min(artifact_size / 10, 10 MiB))`. The artifact `size` attribute is used when available; if it is unknown the fallback is 8 KiB.	`-1`

Yields:

Name	Type	Description
`bytes`	`bytes`	The next chunk of raw artifact bytes.

Example::

# Stream to a file with progress reporting
total = 0
with open("/tmp/output.dat", "wb") as f:
    for chunk in artifact.as_stream():
        f.write(chunk)
        total += len(chunk)
print(f"Downloaded {total} bytes")

Source code in ivcap_client/artifact.py

def as_stream(self, chunk_size: int = -1) -> Iterator[bytes]:
    """Stream the artifact content as a sequence of raw byte chunks.

    Uses an HTTP streaming GET so that only ``chunk_size`` bytes are buffered
    in memory at a time.  This is the lowest-level download method and is
    suitable when you need to:

    * Implement custom progress reporting.
    * Pipe artifact bytes into a third-party streaming API.
    * Process data incrementally without writing a local file.

    For simply saving the artifact to disk, :meth:`as_local_file` is more
    ergonomic.  For loading everything into memory at once, use :meth:`open`.

    Args:
        chunk_size (int): Number of bytes to read per chunk.  Pass ``-1``
            (the default) to let the SDK choose a chunk size automatically:
            ``max(8 KiB, min(artifact_size / 10, 10 MiB))``.  The artifact
            ``size`` attribute is used when available; if it is unknown the
            fallback is 8 KiB.

    Yields:
        bytes: The next chunk of raw artifact bytes.

    Example::

        # Stream to a file with progress reporting
        total = 0
        with open("/tmp/output.dat", "wb") as f:
            for chunk in artifact.as_stream():
                f.write(chunk)
                total += len(chunk)
        print(f"Downloaded {total} bytes")
    """
    chunk_size = _resolve_chunk_size(chunk_size, self.size)
    client = self._ivcap._client.get_httpx_client()
    with client.stream("GET", self._data_href) as response:
        response.raise_for_status()
        for chunk in response.iter_bytes(chunk_size=chunk_size):
            yield chunk

`as_local_file(path=None, chunk_size=-1)` ¶

Download the artifact to a local file and return the path.

This is the recommended method for saving artifact content to disk. It supports two usage patterns depending on whether path is supplied:

Temporary file (path omitted) — a new temp file is created and a :class:CMPath is returned. :class:CMPath is a context-manager-aware :class:~pathlib.Path subclass: when used in a with statement the file is automatically deleted when the block exits. Use this pattern when you only need the file transiently::

with artifact.as_local_file() as path:
    data = path.read_bytes()
# temp file has been deleted here

Explicit path (path provided) — the content is streamed to the given path (parent directories are created automatically). A plain :class:~pathlib.Path is returned; the file is not deleted automatically. Use this pattern when you want to keep the file::

path = artifact.as_local_file("/tmp/output.jpg")
# file remains at /tmp/output.jpg

Parameters:

Name	Type	Description	Default
`path`	`Path \| None`	Destination file path. If `None` (default), a temporary file is created and wrapped in a self-deleting :class:`CMPath`. If a path is provided, the file is written there and a plain :class:`~pathlib.Path` is returned.	`None`
`chunk_size`	`int`	Number of bytes to read per chunk. Pass `-1` (the default) to let the SDK choose automatically based on the artifact size (see :meth:`as_stream` for the formula).	`-1`

Returns:

Type	Description
`Path`	class:`CMPath` when `path` is `None` (context-manager, auto-delete). Plain :class:`~pathlib.Path` when `path` is provided (persists).

Note

Both :class:CMPath and :class:~pathlib.Path are :class:~pathlib.Path subclasses and support all normal path operations (read_bytes(), open(), etc.).

Source code in ivcap_client/artifact.py

def as_local_file(self, path: Path | None = None, chunk_size: int = -1) -> Path:
    """Download the artifact to a local file and return the path.

    This is the **recommended** method for saving artifact content to disk.
    It supports two usage patterns depending on whether ``path`` is supplied:

    **Temporary file (path omitted)** — a new temp file is created and a
    :class:`CMPath` is returned.  :class:`CMPath` is a context-manager-aware
    :class:`~pathlib.Path` subclass: when used in a ``with`` statement the
    file is **automatically deleted** when the block exits.  Use this pattern
    when you only need the file transiently::

        with artifact.as_local_file() as path:
            data = path.read_bytes()
        # temp file has been deleted here

    **Explicit path (path provided)** — the content is streamed to the given
    path (parent directories are created automatically).  A plain
    :class:`~pathlib.Path` is returned; the file is **not** deleted
    automatically.  Use this pattern when you want to keep the file::

        path = artifact.as_local_file("/tmp/output.jpg")
        # file remains at /tmp/output.jpg

    Args:
        path: Destination file path.  If ``None`` (default), a temporary file
            is created and wrapped in a self-deleting :class:`CMPath`.
            If a path is provided, the file is written there and a plain
            :class:`~pathlib.Path` is returned.
        chunk_size (int): Number of bytes to read per chunk.  Pass ``-1``
            (the default) to let the SDK choose automatically based on the
            artifact size (see :meth:`as_stream` for the formula).

    Returns:
        :class:`CMPath` when ``path`` is ``None`` (context-manager, auto-delete).
        Plain :class:`~pathlib.Path` when ``path`` is provided (persists).

    Note:
        Both :class:`CMPath` and :class:`~pathlib.Path` are
        :class:`~pathlib.Path` subclasses and support all normal path
        operations (``read_bytes()``, ``open()``, etc.).
    """
    if path is not None:
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "wb") as f:
            for chunk in self.as_stream(chunk_size=chunk_size):
                f.write(chunk)
        return path
    temp_file = tempfile.NamedTemporaryFile(delete=False)
    with temp_file as f:
        for chunk in self.as_stream(chunk_size=chunk_size):
            f.write(chunk)
    return CMPath(temp_file.name)

`add_metadata(aspect, *, schema=None, policy=None)` ¶

Add a metadata 'aspect' to this artifact. The 'schema' of the aspect, if not defined is expected to found in the 'aspect' under the '$schema' key.

Parameters:

Name	Type	Description	Default
`aspect`	`dict`	The aspect to be attached	required
`schema`	`Optional[str]`	Schema of the aspect. Defaults to 'aspect["$schema"]'.	`None`
`policy`	`URN \| None`	Optional[URN]: Set specific policy controlling access ('urn:ivcap:policy:...').	`None`

Returns:

Name	Type	Description
`self`	`Artifact`	To enable chaining

Source code in ivcap_client/artifact.py

def add_metadata(
    self,
    aspect: dict[str, any],
    *,
    schema: str | None = None,
    policy: URN | None = None,
) -> Artifact:
    """Add a metadata 'aspect' to this artifact. The 'schema' of the aspect, if not defined
    is expected to found in the 'aspect' under the '$schema' key.

    Args:
        aspect (dict): The aspect to be attached
        schema (Optional[str], optional): Schema of the aspect. Defaults to 'aspect["$schema"]'.
        policy: Optional[URN]: Set specific policy controlling access ('urn:ivcap:policy:...').

    Returns:
        self: To enable chaining
    """
    self._ivcap.add_aspect(
        entity=self.id, aspect=aspect, schema=schema, policy=policy
    )
    return self

Artifact¶

Quick Reference¶

Class Documentation¶

Artifact dataclass ¶

open() ¶

as_stream(chunk_size=-1) ¶

as_local_file(path=None, chunk_size=-1) ¶

add_metadata(aspect, *, schema=None, policy=None) ¶

`Artifact` `dataclass` ¶

`open()` ¶

`as_stream(chunk_size=-1)` ¶

`as_local_file(path=None, chunk_size=-1)` ¶

`add_metadata(aspect, *, schema=None, policy=None)` ¶