IVCAP Client¶

The IVCAP class is the single entry point for all client operations. It manages authentication, connection, and exposes all high-level methods for working with services, jobs, artifacts, and aspects.

Quick Reference¶

from ivcap_client.ivcap import IVCAP

# From environment variables (recommended)
ivcap = IVCAP()

# With explicit credentials
ivcap = IVCAP(
    url="https://api.your-ivcap-deployment.net",
    token="<jwt-token>",
    account_id="urn:ivcap:account:<uuid>",
)

Class Documentation¶

`IVCAP` ¶

Entry point for all interactions with an IVCAP deployment.

IVCAP() is the single constructor for all three operating modes. The correct implementation is selected automatically from environment variables — no code changes are needed between local development and deployed operation:

.. code-block:: python

from ivcap_client import IVCAP

ivcap = IVCAP()  # → LocalIVCAP locally, IVCAP on the platform
artifact = ivcap.upload_artifact(name="result.csv", file_path="/tmp/result.csv")

Auto-detection logic (in order):

IVCAP_URL or IVCAP_BASE_URL env var is set → platform :class:IVCAP instance.
url argument is provided → platform :class:IVCAP instance.
token argument provided without a URL → ValueError.
None of the above → :class:LocalIVCAP backed by IVCAP_LOCAL_DIR (default: ivcap-artifacts).

For the full operating-mode reference see :doc:/guides/local-mode.

Source code in ivcap_client/ivcap.py

class IVCAP:
    """Entry point for all interactions with an IVCAP deployment.

    ``IVCAP()`` is the single constructor for all three operating modes.  The
    correct implementation is selected **automatically** from environment
    variables — no code changes are needed between local development and
    deployed operation:

    .. code-block:: python

        from ivcap_client import IVCAP

        ivcap = IVCAP()  # → LocalIVCAP locally, IVCAP on the platform
        artifact = ivcap.upload_artifact(name="result.csv", file_path="/tmp/result.csv")

    **Auto-detection logic (in order):**

    1. ``IVCAP_URL`` or ``IVCAP_BASE_URL`` env var is set → platform
       :class:`IVCAP` instance.
    2. ``url`` argument is provided → platform :class:`IVCAP` instance.
    3. ``token`` argument provided without a URL → ``ValueError``.
    4. None of the above → :class:`LocalIVCAP` backed by ``IVCAP_LOCAL_DIR``
       (default: ``ivcap-artifacts``).

    For the full operating-mode reference see
    :doc:`/guides/local-mode`.
    """

    def __new__(
        cls,
        url: str | None = None,
        token: str | None = None,
        account_id: str | None = None,
    ):
        """Auto-detect platform vs local mode and return the right instance type.

        The decision logic is:

        1. If an explicit ``token`` is passed the caller clearly intends to
           connect to a platform — return a normal :class:`IVCAP` instance and
           let ``__init__`` raise if the URL is also missing.
        2. Otherwise, check for a platform URL (``url`` arg or ``IVCAP_URL`` /
           ``IVCAP_BASE_URL`` env vars).  If one is found, return a normal
           :class:`IVCAP` instance.
        3. If no URL and no explicit token — the environment is unambiguously
           local.  Return a :class:`LocalIVCAP` backed by ``IVCAP_LOCAL_DIR``
           (default: ``ivcap-artifacts``).
        """
        # Explicit token ⇒ platform intent; let __init__ handle validation.
        if token is not None:
            return super().__new__(cls)

        effective_url = (
            url or os.environ.get("IVCAP_URL") or os.environ.get("IVCAP_BASE_URL")
        )
        if not effective_url:
            base_dir = os.environ.get("IVCAP_LOCAL_DIR", "ivcap-artifacts")
            logger.info(
                "No IVCAP platform URL found — using local artifact storage at %r",
                base_dir,
            )
            return LocalIVCAP(base_dir=base_dir)
        # Normal platform instance — Python will call __init__ on this.
        return super().__new__(cls)

    def __init__(
        self,
        url: str | None = None,
        token: str | None = None,
        account_id: str | None = None,
    ):
        """Initialise a platform-connected IVCAP client.

        In normal usage you do not need to call this directly — use ``IVCAP()``
        and let mode auto-detection choose the right implementation.

        Args:
            url (Optional[str]): Base URL of the IVCAP deployment
                (e.g. ``https://api.your-ivcap-deployment.net``).
                Defaults to the ``IVCAP_URL`` environment variable.
                When running inside a platform container, ``IVCAP_BASE_URL``
                is used automatically (no token required in that case).
            token (Optional[str]): JWT bearer token for authentication.
                Required for external (non-container) access.
                Defaults to the ``IVCAP_JWT`` environment variable.
            account_id (Optional[str]): Account URN
                (``urn:ivcap:account:<uuid>``).  Optional — some operations
                (e.g. listing secrets) require it.
                Defaults to the ``IVCAP_ACCOUNT_ID`` environment variable.

        Raises:
            ValueError: If no URL can be determined (neither ``url`` argument
                nor ``IVCAP_URL`` / ``IVCAP_BASE_URL`` env vars are set).
            ValueError: If a URL is set for external access but no JWT token
                can be found (neither ``token`` argument nor ``IVCAP_JWT``
                env var).
        """
        inside_platform = False
        if not url:
            url = os.environ.get("IVCAP_URL")
            if not url:
                url = os.environ.get("IVCAP_BASE_URL")
                inside_platform = url is not None
        if not url:
            # This branch should not be reached in normal use because __new__
            # would have returned a LocalIVCAP already, but guard defensively.
            raise ValueError(
                "missing 'url' argument or environment variables 'IVCAP_URL' or 'IVCAP_BASE_URL' not set."
            )

        if not token:
            token = os.environ.get("IVCAP_JWT")
        self._url = url
        self._token = token
        self._account_id = account_id
        if inside_platform:
            self._client = Client(base_url=url)
        else:
            if not token:
                raise ValueError(
                    "missing 'token' argument or environment variable 'IVCAP_JWT' not set."
                )
            self._client = AuthenticatedClient(base_url=url, token=token)

    #### SERVICES

    def list_services(
        self,
        *,
        filter: str | None = None,
        limit: int | None = 10,
        order_by: str | None = None,
        order_desc: bool | None = False,
        at_time: datetime.datetime | None = UNSET,
    ) -> Iterator[Service]:
        """Return an iterator over all the available services fulfilling certain constraints.

        Args:
            limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                    to be included in the result. Default: 10. Example: 10.
            filter (Optional[str]): The 'filter' system query option allows clients to filter a
                collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                            is evaluated for each resource in the collection, and only items where the expression
                                            evaluates to true are included in the response. Example: name ~= 'Scott%'.
            order_by (Optional[str]): The 'orderby' query option allows clients to request
                resources in either
                                    ascending order using asc or descending order using desc. If asc or desc not specified,
                                    then the resources will be ordered in ascending order. The request below orders Trips on
                                    property EndsAt in descending order. Example: orderby=EndsAt.
            order_desc (Optional[bool]): When set order result in descending order. Ascending
                order is the lt. Default: False.
            at_time (Optional[datetime.datetime]): Return the state of the respective resources at
                that time [now] Example: 1996-12-19T16:39:57-08:00.

        Returns:
            Iterator[Service]: An iterator over a list of services

        Yields:
            Service: A Service object
        """
        kwargs = {
            "filter_": _wrap(filter),
            "limit": _wrap(limit),
            "order_by": _wrap(order_by),
            "order_desc": _wrap(order_desc),
            "at_time": _wrap(at_time),
            "client": self._client,
        }
        return ServiceIter(self, **kwargs)

    def get_service_by_name(self, name: str) -> Service:
        """Return a Service instance named 'name'

        Args:
            name (str): Name of service requested

        Raises:
            ResourceNotFound: Service is not found
            AmbiguousRequest: More than one service is found for 'name'

        Returns:
            Service: The Service instance for the requested service
        """
        l = list(self.list_services(filter=f"name~='{name}'"))
        n = len(l)
        if n == 0:
            raise ResourceNotFound(name)
        elif n > 1:
            raise AmbiguousRequest(f"more than one service '{name} found.")
        return l[0]

    def get_service(self, service_id: URN) -> Service:
        """Returns a Service instance for service 'service_id'

        Args:
            service_id (URN): URN of service

        Returns:
            Service: Returns a Service instance if service exists
        """
        return Service(self, id=service_id)

    ### AGENTS

    def get_agent(self, agent_id: URN) -> Agent:
        """Returns an Agent instance for agent 'agent_id'

        Args:
            agent_id (URN): URN of agent

        Returns:
            Service: Returns an Agent instance if agent exists
        """
        return Agent(self, id=agent_id)

    ### ORDERS

    def list_orders(
        self,
        *,
        filter: str | None = None,
        limit: int | None = 10,
        order_by: str | None = None,
        order_desc: bool | None = False,
        at_time: datetime.datetime | None = UNSET,
    ) -> Iterator[Order]:
        """Return an iterator over all the available orders fulfilling certain constraints.

        Args:
            limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                    to be included in the result. Default: 10. Example: 10.
            filter (Optional[str]): The 'filter' system query option allows clients to filter a
                collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                            is evaluated for each resource in the collection, and only items where the expression
                                            evaluates to true are included in the response. Example: name ~= 'Scott%'.
            order_by (Optional[str]): The 'orderby' query option allows clients to request
                resources in either
                                    ascending order using asc or descending order using desc. If asc or desc not specified,
                                    then the resources will be ordered in ascending order. The request below orders Trips on
                                    property EndsAt in descending order. Example: orderby=EndsAt.
            order_desc (Optional[bool]): When set order result in descending order. Ascending
                order is the lt. Default: False.
            at_time (Optional[datetime.datetime]): Return the state of the respective resources at
                that time [now] Example: 1996-12-19T16:39:57-08:00.

        Returns:
            Iterator[Order]: An iterator over a list of orders

        Yields:
            Order: An order object
        """
        kwargs = {
            "filter_": _wrap(filter),
            "limit": _wrap(limit),
            "order_by": _wrap(order_by),
            "order_desc": _wrap(order_desc),
            "at_time": _wrap(at_time),
            "client": self._client,
        }
        return OrderIter(self, **kwargs)

    def get_order(self, order_id: URN) -> Order:
        """Return an Order instance for the given order URN.

        Args:
            order_id (URN): URN of the order (``urn:ivcap:job:<uuid>``).

        Returns:
            Order: The Order instance for the requested order.
        """
        return Order(self, id=order_id)

    #### ASPECT

    def add_aspect(
        self,
        entity: str,
        aspect: dict[str, any],
        *,
        schema: str | None = None,
        policy: URN | None = None,
    ) -> Aspect:
        """Add an 'aspect' to an 'entity'. The 'schema' of the aspect, if not defined
        is expected to found in the 'aspect' under the '$schema' key.

        Args:
            entity (str): URN of the entity to attach the aspect to
            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:
            aspect: The created aspect record
        """
        return _add_update_aspect(
            self, False, entity, aspect, schema=schema, policy=policy
        )

    def update_aspect(
        self,
        entity: str,
        aspect: dict[str, any],
        *,
        schema: str | None = None,
        policy: URN | None = None,
    ) -> Aspect:
        """Create an 'aspect' to an 'entity', but also retract a
        potentially existing aspect for the same entity with the same schema.
        The 'schema' of the aspect, if not defined
        is expected to found in the 'aspect' under the '$schema' key.

        Args:
            entity (str): URN of the entity to attach the aspect to
            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:
            aspect: The created aspect record
        """
        return _add_update_aspect(
            self, True, entity, aspect, schema=schema, policy=policy
        )

    def list_aspects(
        self,
        *,
        entity: str | None = None,
        schema: str | None = None,
        content_path: str | None = None,
        at_time: datetime.datetime | None = None,
        limit: int | None = 10,
        filter: str | None = None,
        order_by: str | None = "valid_from",
        order_direction: str | None = "DESC",
        include_content: bool | None = True,
    ) -> Iterator[Aspect]:
        """Return an iterator over all the aspect records fulfilling certain constraints.

        Args:
            entity (Optional[str]): Optional entity for which to request aspects Example:
                urn:blue:image.collA#12.
            schema (Optional[str]): Schema prefix using '%' as wildcard indicator Example:
                urn:blue:schema:image%.
            content_path (Optional[str]): To learn more about the supported format, see
                                                    https://www.postgresql.org/docs/current/datatype-json.html#DATATYPE-JSONPATH Example:
                $.images[*] ? (@.size > 10000).
            at_time (Optional[datetime.datetime]): Return aspect which where valid at that time
                [now] Example: 1996-12-19T16:39:57-08:00.
            limit (Optional[int]): The 'limit' system query option requests the number of items in
                the queried
                                            collection to be included in the result. Default: 10. Example: 10.
            filter (Optional[str]): The 'filter' system query option allows clients to filter a collection of
                                            resources that are addressed by a request URL. The expression specified with 'filter'
                                            is evaluated for each resource in the collection, and only items where the expression
                                            evaluates to true are included in the response. Default: ''. Example: FirstName eq
                'Scott'.
            order_by (Optional[str]): Optional comma-separated list of attributes to sort the list
                by.
                * entity
                * schema
                * content
                * policy
                * account
                * created_by
                * retracted_by
                * replaces
                * valid_from
                * valid_to
                Default: 'valid_from'. Example: entity,created-at.
            order_direction (Optional[str]): Set the sort direction 'ASC', 'DESC' for each order-
                by element. Default: 'DESC'. Example: desc.
            include_content (Optional[bool]): When set, also include aspect content in list.

        Returns:
            Iterator[Aspect]: An iterator over a list of aspect records

        Yields:
            Aspect: A aspect object
        """
        kwargs = {
            "entity": _wrap(entity),
            "schema": _wrap(schema),
            "content_path": _wrap(content_path),
            "at_time": _wrap(at_time),
            "limit": _wrap(limit),
            "filter_": _wrap(filter),
            "order_by": _wrap(order_by),
            "order_direction": _wrap(order_direction),
            "include_content": _wrap(include_content),
            "client": self._client,
        }
        return AspectIter(self, **kwargs)

    #### ARTIFACTS

    def list_artifacts(
        self,
        *,
        filter: str | None = None,
        limit: int | None = 10,
        order_by: str | None = None,
        order_desc: bool | None = False,
        at_time: datetime.datetime | None = UNSET,
    ) -> Iterator[Artifact]:
        """Return an iterator over all the available artifacts fulfilling certain constraints.

        Args:
            limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                    to be included in the result. Default: 10. Example: 10.
            filter (Optional[str]): The 'filter' system query option allows clients to filter a
                collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                            is evaluated for each resource in the collection, and only items where the expression
                                            evaluates to true are included in the response. Example: name ~= 'Scott%'.
            order_by (Optional[str]): The 'orderby' query option allows clients to request
                resources in either
                                    ascending order using asc or descending order using desc. If asc or desc not specified,
                                    then the resources will be ordered in ascending order. The request below orders Trips on
                                    property EndsAt in descending order. Example: orderby=EndsAt.
            order_desc (Optional[bool]): When set order result in descending order. Ascending
                order is the lt. Default: False.
            at_time (Optional[datetime.datetime]): Return the state of the respective resources at
                that time [now] Example: 1996-12-19T16:39:57-08:00.

        Returns:
            Iterator[Artifact]: An iterator over a list of artifacts

        Yields:
            Artifact: An artifact object
        """
        kwargs = {
            "filter_": _wrap(filter),
            "limit": _wrap(limit),
            "order_by": _wrap(order_by),
            "order_desc": _wrap(order_desc),
            "at_time": _wrap(at_time),
            "client": self._client,
        }
        return ArtifactIter(self, **kwargs)

    def upload_artifact(
        self,
        *,
        name: str | None = None,
        file_path: str | None = None,
        io_stream: IO | None = None,
        content_type: str | None = None,
        content_size: int | None = -1,
        collection: URN | None = None,
        policy: URN | None = None,
        chunk_size: int | None = MAXSIZE,
        retries: int | None = 0,
        retry_delay: int | None = 30,
        force_upload: bool | None = False,
    ) -> Artifact:
        """Upload a file or byte stream to IVCAP as a new artifact.

        Either ``file_path`` or ``io_stream`` must be provided (not both).
        When using ``io_stream``, ``content_type`` must be supplied explicitly.

        **Deduplication:** When ``file_path`` is given, the SDK stores a hidden
        ``.ivcap-<filename>.txt`` sidecar file next to the source file
        containing an MD5 hash.  On subsequent calls for the same unchanged
        file the existing artifact is returned immediately without re-uploading.
        Override with ``force_upload=True``.

        Args:
            name (Optional[str]): Human-readable display name for the artifact.
            file_path (Optional[str]): Path to the local file to upload.
                The MIME type is auto-detected from the file extension if
                ``content_type`` is not given.
            io_stream (Optional[IO]): In-memory byte stream to upload.
                ``content_type`` must be provided when using this argument.
            content_type (Optional[str]): MIME type of the content.  Required
                when using ``io_stream``; auto-detected from ``file_path``
                extension otherwise.
            content_size (Optional[int]): Size of the content in bytes.
                Defaults to -1 (unknown); auto-determined from ``file_path``
                when possible.
            collection (Optional[URN]): Add the artifact to a named collection
                (``urn:ivcap:collection:<uuid>``).
            policy (Optional[URN]): Access policy URN
                (``urn:ivcap:policy:<name>``).
            chunk_size (Optional[int]): TUS upload chunk size in bytes.
                Defaults to ``sys.maxsize`` (single-chunk upload).
            retries (Optional[int]): Number of retry attempts on upload
                failure.  Defaults to 0 (no retries).
            retry_delay (Optional[int]): Seconds to wait between retries.
                Defaults to 30.
            force_upload (Optional[bool]): Re-upload even if a sidecar file
                indicates the file was already uploaded.  Defaults to False.

        Returns:
            Artifact: The newly created (or previously uploaded) artifact.

        Raises:
            ValueError: If neither ``file_path`` nor ``io_stream`` is
                provided, if the file does not exist or is not readable, or
                if ``content_type`` cannot be determined.

        Example::

            # Upload a local file
            artifact = ivcap.upload_artifact(
                name="my-image",
                file_path="/path/to/image.jpg",
            )

            # Upload from an in-memory stream
            import io
            data = b"col1,col2\\n1,2\\n3,4\\n"
            artifact = ivcap.upload_artifact(
                name="my-data.csv",
                io_stream=io.BytesIO(data),
                content_type="text/csv",
                content_size=len(data),
            )
        """
        from ivcap_client.artifact import upload_artifact as upload

        return upload(
            self,
            name=name,
            file_path=file_path,
            io_stream=io_stream,
            content_type=content_type,
            content_size=content_size,
            collection=collection,
            policy=policy,
            chunk_size=chunk_size,
            retries=retries,
            retry_delay=retry_delay,
            force_upload=force_upload,
        )

    def artifact_for_file(self, file_path: str) -> Artifact | None:
        """Return an Artifact instance if local file 'file_path'
        has already been uploaded as artifact.

        Args:
            file_path (str): Path to local file

        Returns:
            Optional[Artifact]: Return artifact instance if file has been uploaded,
            otherwise return None
        """
        aurn = check_file_already_uploaded(file_path)
        if aurn is not None:
            return self.get_artifact(aurn)

    def get_artifact(self, id: URN) -> Artifact:
        """Returns an Artifact instance for artifact 'id'

        Args:
            id (URN): URN of artifact

        Returns:
            Artifact: Returns an Artifact instance if artifact exists
        """
        if id.startswith("file://") or id.startswith("urn:file://"):
            return LocalFileArtifact(id)
        return Artifact(self, id=id).refresh()

    #### COLLECTIONS

    def create_collection(
        self,
        urn: str,
        name: str,
        *,
        description: str | None = None,
        policy: URN | None = None,
    ) -> Collection:
        """Create or update a collection definition (idempotent via PUT).

        Calling this method on an already-existing collection URN **replaces**
        the previous name/description without affecting its items.

        Args:
            urn (str): The collection entity URN
                (e.g. ``urn:ivcap:collection:<uuid>``).
            name (str): Human-readable collection name.
            description (Optional[str]): Optional description.
            policy (Optional[URN]): Access policy URN
                (``urn:ivcap:policy:…``).

        Returns:
            Collection: The created or updated collection.
        """
        return create_collection(
            self, urn, name, description=description, policy=policy
        )

    def get_collection(
        self,
        urn: str,
        *,
        at_time: datetime | None = None,
    ) -> Collection:
        """Fetch a collection definition by its URN.

        Args:
            urn (str): The collection entity URN.
            at_time (Optional[datetime]): Retrieve the state at this point in
                time.

        Returns:
            Collection: The collection instance.

        Raises:
            ResourceNotFound: If no collection with the given URN exists.
        """
        return get_collection(self, urn, at_time=at_time)

    def list_collections(
        self,
        *,
        name_filter: str | None = None,
        limit: int | None = 10,
        at_time: datetime | None = None,
    ) -> Iterator[Collection]:
        """Return an iterator over collection definitions.

        Args:
            name_filter (Optional[str]): A JSONPath comparison expression
                applied to the collection ``name`` field.  The expression is
                wrapped as ``$.name ? (@ <name_filter>)`` before being sent
                to the server.

                Examples::

                    '== "My Ocean Survey"'
                    'starts with "CTD"'
                    'like_regex ".*ocean.*" flag "i"'

            limit (Optional[int]): Maximum number of collections to return.
                Default: 10.
            at_time (Optional[datetime]): Return collections valid at this
                point in time.

        Returns:
            Iterator[Collection]: An iterator over collections.
        """
        return list_collections(
            self, name_filter=name_filter, limit=limit, at_time=at_time
        )

    def add_to_collection(
        self,
        collection_urn: str,
        item_urn: str,
        *,
        policy: URN | None = None,
    ) -> CollectionItem | None:
        """Add an item to a collection with automatic deduplication.

        Checks whether *item_urn* is already a member before creating the
        membership aspect.  If it is already present, returns ``None``
        (skip silently).

        Args:
            collection_urn (str): The collection entity URN.
            item_urn (str): URN of the entity to add.
            policy (Optional[URN]): Optional access policy URN for the
                membership aspect (``urn:ivcap:policy:…``).

        Returns:
            CollectionItem if the item was newly added, ``None`` if it was
            already a member.
        """
        return add_item_to_collection(self, collection_urn, item_urn, policy=policy)

    def remove_from_collection(
        self,
        collection_urn: str,
        item_urn: str,
    ) -> bool:
        """Remove an item from a collection by retracting its membership aspect.

        Items that are not currently members of the collection are silently
        skipped.

        Args:
            collection_urn (str): The collection entity URN.
            item_urn (str): URN of the entity to remove.

        Returns:
            ``True`` if the membership aspect was retracted, ``False`` if the
            item was not a member.
        """
        return remove_item_from_collection(self, collection_urn, item_urn)

    def retract_collection(
        self,
        collection_urn: str,
    ) -> int:
        """Fully retract a collection and all its item memberships.

        All membership aspects are retracted first (paginated), then the
        collection definition aspect is retracted.  This operation cannot
        be undone.

        Args:
            collection_urn (str): URN of the collection to retract.

        Returns:
            Total number of aspect records retracted (items + 1 definition).

        Raises:
            ResourceNotFound: If no collection definition exists for the URN.
        """
        return retract_collection(self, collection_urn)

    #### SECRETS

    def list_secrets(
        self,
        *,
        filter: str | None = None,
        limit: int | None = 10,
        order_by: str | None = None,
        order_desc: bool | None = False,
        at_time: datetime.datetime | None = UNSET,
    ) -> Iterator[Secret]:
        """Return an iterator over all the available secrets fulfilling certain constraints.

        Args:
            limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                    to be included in the result. Default: 10. Example: 10.
            filter (Optional[str]): The 'filter' system query option allows clients to filter a
                collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                            is evaluated for each resource in the collection, and only items where the expression
                                            evaluates to true are included in the response. Example: name ~= 'Scott%'.

        Returns:
            Iterator[Secret]: An iterator over a list of secrets

        Yields:
            Secret: A secret object
        """
        kwargs = {
            "filter_": _wrap(filter),
            "limit": _wrap(limit),
            "client": self._client,
        }
        return SecretIter(self, **kwargs)

    #### SEARCH

    def search(self, query: str) -> Any:
        """Execute query provided in body and return a list of search result.

        Args:
            query: The search query to execute.

        Raises:
            errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
            httpx.TimeoutException: If the request takes longer than Client.timeout.

        Returns:
            Search results.

        """
        body = File(query)
        kwargs = {
            "body": body,
            "content_type": "application/datalog+mangle",
        }
        r = search_search.sync_detailed(client=self._client, **kwargs)
        if r.status_code >= 300:
            return process_error("search", r)
        return r.parsed

    @classmethod
    def local(cls, base_dir: str | None = None) -> LocalIVCAP:
        """Return a :class:`LocalIVCAP` instance for filesystem-only
        (no-network) development and testing.

        This is the preferred way to force local mode regardless of which
        environment variables are set — for example in unit tests:

        .. code-block:: python

            from ivcap_client import IVCAP

            ivcap = IVCAP.local(base_dir="./my-artifacts")
            artifact = ivcap.upload_artifact(name="result.csv", file_path="/tmp/result.csv")

        The ``base_dir`` can also be provided via the ``IVCAP_LOCAL_DIR``
        environment variable.  Precedence (highest first):

        1. The ``base_dir`` argument to this method.
        2. The ``IVCAP_LOCAL_DIR`` environment variable.
        3. The default ``"ivcap-artifacts"``.

        Args:
            base_dir: Root directory for artifact and aspect storage.
                Created on demand.  Defaults to ``IVCAP_LOCAL_DIR`` env var,
                or ``"ivcap-artifacts"`` if neither is set.

        Returns:
            LocalIVCAP: A filesystem-backed client with the same
            ``upload_artifact`` / ``get_artifact`` / ``add_aspect`` interface
            as :class:`IVCAP`.
        """
        if base_dir is None:
            base_dir = os.environ.get("IVCAP_LOCAL_DIR", "ivcap-artifacts")
        return LocalIVCAP(base_dir=base_dir)

    @property
    def url(self) -> str:
        """Returns the URL of the IVCAP deployment

        Returns:
            str: URL of IVCAP deployment
        """
        return self._url

    def __repr__(self):
        return f"<IVCAP url={self._url}>"

`url` `property` ¶

Returns the URL of the IVCAP deployment

Returns:

Name	Type	Description
`str`	`str`	URL of IVCAP deployment

`list_services(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

Return an iterator over all the available services fulfilling certain constraints.

Parameters:

Name	Type	Description	Default
`limit`	`Optional[int]`	The 'limit' query option sets the maximum number of items to be included in the result. Default: 10. Example: 10.	`10`
`filter`	`Optional[str]`	The 'filter' system query option allows clients to filter a collection of resources that are addressed by a request URL. The expression specified with 'filter' is evaluated for each resource in the collection, and only items where the expression evaluates to true are included in the response. Example: name ~= 'Scott%'.	`None`
`order_by`	`Optional[str]`	The 'orderby' query option allows clients to request resources in either ascending order using asc or descending order using desc. If asc or desc not specified, then the resources will be ordered in ascending order. The request below orders Trips on property EndsAt in descending order. Example: orderby=EndsAt.	`None`
`order_desc`	`Optional[bool]`	When set order result in descending order. Ascending order is the lt. Default: False.	`False`
`at_time`	`Optional[datetime]`	Return the state of the respective resources at that time [now] Example: 1996-12-19T16:39:57-08:00.	`UNSET`

Returns:

Type	Description
`Iterator[Service]`	Iterator[Service]: An iterator over a list of services

Yields:

Name	Type	Description
`Service`	`Service`	A Service object

Source code in ivcap_client/ivcap.py

def list_services(
    self,
    *,
    filter: str | None = None,
    limit: int | None = 10,
    order_by: str | None = None,
    order_desc: bool | None = False,
    at_time: datetime.datetime | None = UNSET,
) -> Iterator[Service]:
    """Return an iterator over all the available services fulfilling certain constraints.

    Args:
        limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                to be included in the result. Default: 10. Example: 10.
        filter (Optional[str]): The 'filter' system query option allows clients to filter a
            collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                        is evaluated for each resource in the collection, and only items where the expression
                                        evaluates to true are included in the response. Example: name ~= 'Scott%'.
        order_by (Optional[str]): The 'orderby' query option allows clients to request
            resources in either
                                ascending order using asc or descending order using desc. If asc or desc not specified,
                                then the resources will be ordered in ascending order. The request below orders Trips on
                                property EndsAt in descending order. Example: orderby=EndsAt.
        order_desc (Optional[bool]): When set order result in descending order. Ascending
            order is the lt. Default: False.
        at_time (Optional[datetime.datetime]): Return the state of the respective resources at
            that time [now] Example: 1996-12-19T16:39:57-08:00.

    Returns:
        Iterator[Service]: An iterator over a list of services

    Yields:
        Service: A Service object
    """
    kwargs = {
        "filter_": _wrap(filter),
        "limit": _wrap(limit),
        "order_by": _wrap(order_by),
        "order_desc": _wrap(order_desc),
        "at_time": _wrap(at_time),
        "client": self._client,
    }
    return ServiceIter(self, **kwargs)

`get_service_by_name(name)` ¶

Return a Service instance named 'name'

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of service requested	required

Raises:

Type	Description
`ResourceNotFound`	Service is not found
`AmbiguousRequest`	More than one service is found for 'name'

Returns:

Name	Type	Description
`Service`	`Service`	The Service instance for the requested service

Source code in ivcap_client/ivcap.py

def get_service_by_name(self, name: str) -> Service:
    """Return a Service instance named 'name'

    Args:
        name (str): Name of service requested

    Raises:
        ResourceNotFound: Service is not found
        AmbiguousRequest: More than one service is found for 'name'

    Returns:
        Service: The Service instance for the requested service
    """
    l = list(self.list_services(filter=f"name~='{name}'"))
    n = len(l)
    if n == 0:
        raise ResourceNotFound(name)
    elif n > 1:
        raise AmbiguousRequest(f"more than one service '{name} found.")
    return l[0]

`get_service(service_id)` ¶

Returns a Service instance for service 'service_id'

Parameters:

Name	Type	Description	Default
`service_id`	`URN`	URN of service	required

Returns:

Name	Type	Description
`Service`	`Service`	Returns a Service instance if service exists

Source code in ivcap_client/ivcap.py

def get_service(self, service_id: URN) -> Service:
    """Returns a Service instance for service 'service_id'

    Args:
        service_id (URN): URN of service

    Returns:
        Service: Returns a Service instance if service exists
    """
    return Service(self, id=service_id)

`get_agent(agent_id)` ¶

Returns an Agent instance for agent 'agent_id'

Parameters:

Name	Type	Description	Default
`agent_id`	`URN`	URN of agent	required

Returns:

Name	Type	Description
`Service`	`Agent`	Returns an Agent instance if agent exists

Source code in ivcap_client/ivcap.py

def get_agent(self, agent_id: URN) -> Agent:
    """Returns an Agent instance for agent 'agent_id'

    Args:
        agent_id (URN): URN of agent

    Returns:
        Service: Returns an Agent instance if agent exists
    """
    return Agent(self, id=agent_id)

`list_orders(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

Return an iterator over all the available orders fulfilling certain constraints.

Parameters:

Name	Type	Description	Default
`limit`	`Optional[int]`	The 'limit' query option sets the maximum number of items to be included in the result. Default: 10. Example: 10.	`10`
`filter`	`Optional[str]`	The 'filter' system query option allows clients to filter a collection of resources that are addressed by a request URL. The expression specified with 'filter' is evaluated for each resource in the collection, and only items where the expression evaluates to true are included in the response. Example: name ~= 'Scott%'.	`None`
`order_by`	`Optional[str]`	The 'orderby' query option allows clients to request resources in either ascending order using asc or descending order using desc. If asc or desc not specified, then the resources will be ordered in ascending order. The request below orders Trips on property EndsAt in descending order. Example: orderby=EndsAt.	`None`
`order_desc`	`Optional[bool]`	When set order result in descending order. Ascending order is the lt. Default: False.	`False`
`at_time`	`Optional[datetime]`	Return the state of the respective resources at that time [now] Example: 1996-12-19T16:39:57-08:00.	`UNSET`

Returns:

Type	Description
`Iterator[Order]`	Iterator[Order]: An iterator over a list of orders

Yields:

Name	Type	Description
`Order`	`Order`	An order object

Source code in ivcap_client/ivcap.py

def list_orders(
    self,
    *,
    filter: str | None = None,
    limit: int | None = 10,
    order_by: str | None = None,
    order_desc: bool | None = False,
    at_time: datetime.datetime | None = UNSET,
) -> Iterator[Order]:
    """Return an iterator over all the available orders fulfilling certain constraints.

    Args:
        limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                to be included in the result. Default: 10. Example: 10.
        filter (Optional[str]): The 'filter' system query option allows clients to filter a
            collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                        is evaluated for each resource in the collection, and only items where the expression
                                        evaluates to true are included in the response. Example: name ~= 'Scott%'.
        order_by (Optional[str]): The 'orderby' query option allows clients to request
            resources in either
                                ascending order using asc or descending order using desc. If asc or desc not specified,
                                then the resources will be ordered in ascending order. The request below orders Trips on
                                property EndsAt in descending order. Example: orderby=EndsAt.
        order_desc (Optional[bool]): When set order result in descending order. Ascending
            order is the lt. Default: False.
        at_time (Optional[datetime.datetime]): Return the state of the respective resources at
            that time [now] Example: 1996-12-19T16:39:57-08:00.

    Returns:
        Iterator[Order]: An iterator over a list of orders

    Yields:
        Order: An order object
    """
    kwargs = {
        "filter_": _wrap(filter),
        "limit": _wrap(limit),
        "order_by": _wrap(order_by),
        "order_desc": _wrap(order_desc),
        "at_time": _wrap(at_time),
        "client": self._client,
    }
    return OrderIter(self, **kwargs)

`get_order(order_id)` ¶

Return an Order instance for the given order URN.

Parameters:

Name	Type	Description	Default
`order_id`	`URN`	URN of the order (`urn:ivcap:job:<uuid>`).	required

Returns:

Name	Type	Description
`Order`	`Order`	The Order instance for the requested order.

Source code in ivcap_client/ivcap.py

def get_order(self, order_id: URN) -> Order:
    """Return an Order instance for the given order URN.

    Args:
        order_id (URN): URN of the order (``urn:ivcap:job:<uuid>``).

    Returns:
        Order: The Order instance for the requested order.
    """
    return Order(self, id=order_id)

`add_aspect(entity, aspect, *, schema=None, policy=None)` ¶

Add an 'aspect' to an 'entity'. The 'schema' of the aspect, if not defined is expected to found in the 'aspect' under the '$schema' key.

Parameters:

Name	Type	Description	Default
`entity`	`str`	URN of the entity to attach the aspect to	required
`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
`aspect`	`Aspect`	The created aspect record

Source code in ivcap_client/ivcap.py

def add_aspect(
    self,
    entity: str,
    aspect: dict[str, any],
    *,
    schema: str | None = None,
    policy: URN | None = None,
) -> Aspect:
    """Add an 'aspect' to an 'entity'. The 'schema' of the aspect, if not defined
    is expected to found in the 'aspect' under the '$schema' key.

    Args:
        entity (str): URN of the entity to attach the aspect to
        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:
        aspect: The created aspect record
    """
    return _add_update_aspect(
        self, False, entity, aspect, schema=schema, policy=policy
    )

`update_aspect(entity, aspect, *, schema=None, policy=None)` ¶

Create an 'aspect' to an 'entity', but also retract a potentially existing aspect for the same entity with the same schema. The 'schema' of the aspect, if not defined is expected to found in the 'aspect' under the '$schema' key.

Parameters:

Name	Type	Description	Default
`entity`	`str`	URN of the entity to attach the aspect to	required
`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
`aspect`	`Aspect`	The created aspect record

Source code in ivcap_client/ivcap.py

def update_aspect(
    self,
    entity: str,
    aspect: dict[str, any],
    *,
    schema: str | None = None,
    policy: URN | None = None,
) -> Aspect:
    """Create an 'aspect' to an 'entity', but also retract a
    potentially existing aspect for the same entity with the same schema.
    The 'schema' of the aspect, if not defined
    is expected to found in the 'aspect' under the '$schema' key.

    Args:
        entity (str): URN of the entity to attach the aspect to
        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:
        aspect: The created aspect record
    """
    return _add_update_aspect(
        self, True, entity, aspect, schema=schema, policy=policy
    )

`list_aspects(*, entity=None, schema=None, content_path=None, at_time=None, limit=10, filter=None, order_by='valid_from', order_direction='DESC', include_content=True)` ¶

Return an iterator over all the aspect records fulfilling certain constraints.

Parameters:

Name	Type	Description	Default
`entity`	`Optional[str]`	Optional entity for which to request aspects Example: urn:blue:image.collA#12.	`None`
`schema`	`Optional[str]`	Schema prefix using '%' as wildcard indicator Example: urn:blue:schema:image%.	`None`
`content_path`	`Optional[str]`	To learn more about the supported format, see https://www.postgresql.org/docs/current/datatype-json.html#DATATYPE-JSONPATH Example: $.images[*] ? (@.size > 10000).	`None`
`at_time`	`Optional[datetime]`	Return aspect which where valid at that time [now] Example: 1996-12-19T16:39:57-08:00.	`None`
`limit`	`Optional[int]`	The 'limit' system query option requests the number of items in the queried collection to be included in the result. Default: 10. Example: 10.	`10`
`filter`	`Optional[str]`	The 'filter' system query option allows clients to filter a collection of resources that are addressed by a request URL. The expression specified with 'filter' is evaluated for each resource in the collection, and only items where the expression evaluates to true are included in the response. Default: ''. Example: FirstName eq 'Scott'.	`None`
`order_by`	`Optional[str]`	Optional comma-separated list of attributes to sort the list by. * entity * schema * content * policy * account * created_by * retracted_by * replaces * valid_from * valid_to Default: 'valid_from'. Example: entity,created-at.	`'valid_from'`
`order_direction`	`Optional[str]`	Set the sort direction 'ASC', 'DESC' for each order- by element. Default: 'DESC'. Example: desc.	`'DESC'`
`include_content`	`Optional[bool]`	When set, also include aspect content in list.	`True`

Returns:

Type	Description
`Iterator[Aspect]`	Iterator[Aspect]: An iterator over a list of aspect records

Yields:

Name	Type	Description
`Aspect`	`Aspect`	A aspect object

Source code in ivcap_client/ivcap.py

def list_aspects(
    self,
    *,
    entity: str | None = None,
    schema: str | None = None,
    content_path: str | None = None,
    at_time: datetime.datetime | None = None,
    limit: int | None = 10,
    filter: str | None = None,
    order_by: str | None = "valid_from",
    order_direction: str | None = "DESC",
    include_content: bool | None = True,
) -> Iterator[Aspect]:
    """Return an iterator over all the aspect records fulfilling certain constraints.

    Args:
        entity (Optional[str]): Optional entity for which to request aspects Example:
            urn:blue:image.collA#12.
        schema (Optional[str]): Schema prefix using '%' as wildcard indicator Example:
            urn:blue:schema:image%.
        content_path (Optional[str]): To learn more about the supported format, see
                                                https://www.postgresql.org/docs/current/datatype-json.html#DATATYPE-JSONPATH Example:
            $.images[*] ? (@.size > 10000).
        at_time (Optional[datetime.datetime]): Return aspect which where valid at that time
            [now] Example: 1996-12-19T16:39:57-08:00.
        limit (Optional[int]): The 'limit' system query option requests the number of items in
            the queried
                                        collection to be included in the result. Default: 10. Example: 10.
        filter (Optional[str]): The 'filter' system query option allows clients to filter a collection of
                                        resources that are addressed by a request URL. The expression specified with 'filter'
                                        is evaluated for each resource in the collection, and only items where the expression
                                        evaluates to true are included in the response. Default: ''. Example: FirstName eq
            'Scott'.
        order_by (Optional[str]): Optional comma-separated list of attributes to sort the list
            by.
            * entity
            * schema
            * content
            * policy
            * account
            * created_by
            * retracted_by
            * replaces
            * valid_from
            * valid_to
            Default: 'valid_from'. Example: entity,created-at.
        order_direction (Optional[str]): Set the sort direction 'ASC', 'DESC' for each order-
            by element. Default: 'DESC'. Example: desc.
        include_content (Optional[bool]): When set, also include aspect content in list.

    Returns:
        Iterator[Aspect]: An iterator over a list of aspect records

    Yields:
        Aspect: A aspect object
    """
    kwargs = {
        "entity": _wrap(entity),
        "schema": _wrap(schema),
        "content_path": _wrap(content_path),
        "at_time": _wrap(at_time),
        "limit": _wrap(limit),
        "filter_": _wrap(filter),
        "order_by": _wrap(order_by),
        "order_direction": _wrap(order_direction),
        "include_content": _wrap(include_content),
        "client": self._client,
    }
    return AspectIter(self, **kwargs)

`list_artifacts(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

Return an iterator over all the available artifacts fulfilling certain constraints.

Parameters:

Name	Type	Description	Default
`limit`	`Optional[int]`	The 'limit' query option sets the maximum number of items to be included in the result. Default: 10. Example: 10.	`10`
`filter`	`Optional[str]`	The 'filter' system query option allows clients to filter a collection of resources that are addressed by a request URL. The expression specified with 'filter' is evaluated for each resource in the collection, and only items where the expression evaluates to true are included in the response. Example: name ~= 'Scott%'.	`None`
`order_by`	`Optional[str]`	The 'orderby' query option allows clients to request resources in either ascending order using asc or descending order using desc. If asc or desc not specified, then the resources will be ordered in ascending order. The request below orders Trips on property EndsAt in descending order. Example: orderby=EndsAt.	`None`
`order_desc`	`Optional[bool]`	When set order result in descending order. Ascending order is the lt. Default: False.	`False`
`at_time`	`Optional[datetime]`	Return the state of the respective resources at that time [now] Example: 1996-12-19T16:39:57-08:00.	`UNSET`

Returns:

Type	Description
`Iterator[Artifact]`	Iterator[Artifact]: An iterator over a list of artifacts

Yields:

Name	Type	Description
`Artifact`	`Artifact`	An artifact object

Source code in ivcap_client/ivcap.py

def list_artifacts(
    self,
    *,
    filter: str | None = None,
    limit: int | None = 10,
    order_by: str | None = None,
    order_desc: bool | None = False,
    at_time: datetime.datetime | None = UNSET,
) -> Iterator[Artifact]:
    """Return an iterator over all the available artifacts fulfilling certain constraints.

    Args:
        limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                to be included in the result. Default: 10. Example: 10.
        filter (Optional[str]): The 'filter' system query option allows clients to filter a
            collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                        is evaluated for each resource in the collection, and only items where the expression
                                        evaluates to true are included in the response. Example: name ~= 'Scott%'.
        order_by (Optional[str]): The 'orderby' query option allows clients to request
            resources in either
                                ascending order using asc or descending order using desc. If asc or desc not specified,
                                then the resources will be ordered in ascending order. The request below orders Trips on
                                property EndsAt in descending order. Example: orderby=EndsAt.
        order_desc (Optional[bool]): When set order result in descending order. Ascending
            order is the lt. Default: False.
        at_time (Optional[datetime.datetime]): Return the state of the respective resources at
            that time [now] Example: 1996-12-19T16:39:57-08:00.

    Returns:
        Iterator[Artifact]: An iterator over a list of artifacts

    Yields:
        Artifact: An artifact object
    """
    kwargs = {
        "filter_": _wrap(filter),
        "limit": _wrap(limit),
        "order_by": _wrap(order_by),
        "order_desc": _wrap(order_desc),
        "at_time": _wrap(at_time),
        "client": self._client,
    }
    return ArtifactIter(self, **kwargs)

`upload_artifact(*, name=None, file_path=None, io_stream=None, content_type=None, content_size=-1, collection=None, policy=None, chunk_size=MAXSIZE, retries=0, retry_delay=30, force_upload=False)` ¶

Upload a file or byte stream to IVCAP as a new artifact.

Either file_path or io_stream must be provided (not both). When using io_stream, content_type must be supplied explicitly.

Deduplication: When file_path is given, the SDK stores a hidden .ivcap-<filename>.txt sidecar file next to the source file containing an MD5 hash. On subsequent calls for the same unchanged file the existing artifact is returned immediately without re-uploading. Override with force_upload=True.

Parameters:

Name	Type	Description	Default
`name`	`Optional[str]`	Human-readable display name for the artifact.	`None`
`file_path`	`Optional[str]`	Path to the local file to upload. The MIME type is auto-detected from the file extension if `content_type` is not given.	`None`
`io_stream`	`Optional[IO]`	In-memory byte stream to upload. `content_type` must be provided when using this argument.	`None`
`content_type`	`Optional[str]`	MIME type of the content. Required when using `io_stream`; auto-detected from `file_path` extension otherwise.	`None`
`content_size`	`Optional[int]`	Size of the content in bytes. Defaults to -1 (unknown); auto-determined from `file_path` when possible.	`-1`
`collection`	`Optional[URN]`	Add the artifact to a named collection (`urn:ivcap:collection:<uuid>`).	`None`
`policy`	`Optional[URN]`	Access policy URN (`urn:ivcap:policy:<name>`).	`None`
`chunk_size`	`Optional[int]`	TUS upload chunk size in bytes. Defaults to `sys.maxsize` (single-chunk upload).	`maxsize`
`retries`	`Optional[int]`	Number of retry attempts on upload failure. Defaults to 0 (no retries).	`0`
`retry_delay`	`Optional[int]`	Seconds to wait between retries. Defaults to 30.	`30`
`force_upload`	`Optional[bool]`	Re-upload even if a sidecar file indicates the file was already uploaded. Defaults to False.	`False`

Returns:

Name	Type	Description
`Artifact`	`Artifact`	The newly created (or previously uploaded) artifact.

Raises:

Type	Description
`ValueError`	If neither `file_path` nor `io_stream` is provided, if the file does not exist or is not readable, or if `content_type` cannot be determined.

Example::

# Upload a local file
artifact = ivcap.upload_artifact(
    name="my-image",
    file_path="/path/to/image.jpg",
)

# Upload from an in-memory stream
import io
data = b"col1,col2\n1,2\n3,4\n"
artifact = ivcap.upload_artifact(
    name="my-data.csv",
    io_stream=io.BytesIO(data),
    content_type="text/csv",
    content_size=len(data),
)

Source code in ivcap_client/ivcap.py

def upload_artifact(
    self,
    *,
    name: str | None = None,
    file_path: str | None = None,
    io_stream: IO | None = None,
    content_type: str | None = None,
    content_size: int | None = -1,
    collection: URN | None = None,
    policy: URN | None = None,
    chunk_size: int | None = MAXSIZE,
    retries: int | None = 0,
    retry_delay: int | None = 30,
    force_upload: bool | None = False,
) -> Artifact:
    """Upload a file or byte stream to IVCAP as a new artifact.

    Either ``file_path`` or ``io_stream`` must be provided (not both).
    When using ``io_stream``, ``content_type`` must be supplied explicitly.

    **Deduplication:** When ``file_path`` is given, the SDK stores a hidden
    ``.ivcap-<filename>.txt`` sidecar file next to the source file
    containing an MD5 hash.  On subsequent calls for the same unchanged
    file the existing artifact is returned immediately without re-uploading.
    Override with ``force_upload=True``.

    Args:
        name (Optional[str]): Human-readable display name for the artifact.
        file_path (Optional[str]): Path to the local file to upload.
            The MIME type is auto-detected from the file extension if
            ``content_type`` is not given.
        io_stream (Optional[IO]): In-memory byte stream to upload.
            ``content_type`` must be provided when using this argument.
        content_type (Optional[str]): MIME type of the content.  Required
            when using ``io_stream``; auto-detected from ``file_path``
            extension otherwise.
        content_size (Optional[int]): Size of the content in bytes.
            Defaults to -1 (unknown); auto-determined from ``file_path``
            when possible.
        collection (Optional[URN]): Add the artifact to a named collection
            (``urn:ivcap:collection:<uuid>``).
        policy (Optional[URN]): Access policy URN
            (``urn:ivcap:policy:<name>``).
        chunk_size (Optional[int]): TUS upload chunk size in bytes.
            Defaults to ``sys.maxsize`` (single-chunk upload).
        retries (Optional[int]): Number of retry attempts on upload
            failure.  Defaults to 0 (no retries).
        retry_delay (Optional[int]): Seconds to wait between retries.
            Defaults to 30.
        force_upload (Optional[bool]): Re-upload even if a sidecar file
            indicates the file was already uploaded.  Defaults to False.

    Returns:
        Artifact: The newly created (or previously uploaded) artifact.

    Raises:
        ValueError: If neither ``file_path`` nor ``io_stream`` is
            provided, if the file does not exist or is not readable, or
            if ``content_type`` cannot be determined.

    Example::

        # Upload a local file
        artifact = ivcap.upload_artifact(
            name="my-image",
            file_path="/path/to/image.jpg",
        )

        # Upload from an in-memory stream
        import io
        data = b"col1,col2\\n1,2\\n3,4\\n"
        artifact = ivcap.upload_artifact(
            name="my-data.csv",
            io_stream=io.BytesIO(data),
            content_type="text/csv",
            content_size=len(data),
        )
    """
    from ivcap_client.artifact import upload_artifact as upload

    return upload(
        self,
        name=name,
        file_path=file_path,
        io_stream=io_stream,
        content_type=content_type,
        content_size=content_size,
        collection=collection,
        policy=policy,
        chunk_size=chunk_size,
        retries=retries,
        retry_delay=retry_delay,
        force_upload=force_upload,
    )

`artifact_for_file(file_path)` ¶

Return an Artifact instance if local file 'file_path' has already been uploaded as artifact.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Path to local file	required

Returns:

Type	Description
`Artifact \| None`	Optional[Artifact]: Return artifact instance if file has been uploaded, otherwise return None

Source code in ivcap_client/ivcap.py

def artifact_for_file(self, file_path: str) -> Artifact | None:
    """Return an Artifact instance if local file 'file_path'
    has already been uploaded as artifact.

    Args:
        file_path (str): Path to local file

    Returns:
        Optional[Artifact]: Return artifact instance if file has been uploaded,
        otherwise return None
    """
    aurn = check_file_already_uploaded(file_path)
    if aurn is not None:
        return self.get_artifact(aurn)

`get_artifact(id)` ¶

Returns an Artifact instance for artifact 'id'

Parameters:

Name	Type	Description	Default
`id`	`URN`	URN of artifact	required

Returns:

Name	Type	Description
`Artifact`	`Artifact`	Returns an Artifact instance if artifact exists

Source code in ivcap_client/ivcap.py

def get_artifact(self, id: URN) -> Artifact:
    """Returns an Artifact instance for artifact 'id'

    Args:
        id (URN): URN of artifact

    Returns:
        Artifact: Returns an Artifact instance if artifact exists
    """
    if id.startswith("file://") or id.startswith("urn:file://"):
        return LocalFileArtifact(id)
    return Artifact(self, id=id).refresh()

`create_collection(urn, name, *, description=None, policy=None)` ¶

Create or update a collection definition (idempotent via PUT).

Calling this method on an already-existing collection URN replaces the previous name/description without affecting its items.

Parameters:

Name	Type	Description	Default
`urn`	`str`	The collection entity URN (e.g. `urn:ivcap:collection:<uuid>`).	required
`name`	`str`	Human-readable collection name.	required
`description`	`Optional[str]`	Optional description.	`None`
`policy`	`Optional[URN]`	Access policy URN (`urn:ivcap:policy:…`).	`None`

Returns:

Name	Type	Description
`Collection`	`Collection`	The created or updated collection.

Source code in ivcap_client/ivcap.py

def create_collection(
    self,
    urn: str,
    name: str,
    *,
    description: str | None = None,
    policy: URN | None = None,
) -> Collection:
    """Create or update a collection definition (idempotent via PUT).

    Calling this method on an already-existing collection URN **replaces**
    the previous name/description without affecting its items.

    Args:
        urn (str): The collection entity URN
            (e.g. ``urn:ivcap:collection:<uuid>``).
        name (str): Human-readable collection name.
        description (Optional[str]): Optional description.
        policy (Optional[URN]): Access policy URN
            (``urn:ivcap:policy:…``).

    Returns:
        Collection: The created or updated collection.
    """
    return create_collection(
        self, urn, name, description=description, policy=policy
    )

`get_collection(urn, *, at_time=None)` ¶

Fetch a collection definition by its URN.

Parameters:

Name	Type	Description	Default
`urn`	`str`	The collection entity URN.	required
`at_time`	`Optional[datetime]`	Retrieve the state at this point in time.	`None`

Returns:

Name	Type	Description
`Collection`	`Collection`	The collection instance.

Raises:

Type	Description
`ResourceNotFound`	If no collection with the given URN exists.

Source code in ivcap_client/ivcap.py

def get_collection(
    self,
    urn: str,
    *,
    at_time: datetime | None = None,
) -> Collection:
    """Fetch a collection definition by its URN.

    Args:
        urn (str): The collection entity URN.
        at_time (Optional[datetime]): Retrieve the state at this point in
            time.

    Returns:
        Collection: The collection instance.

    Raises:
        ResourceNotFound: If no collection with the given URN exists.
    """
    return get_collection(self, urn, at_time=at_time)

`list_collections(*, name_filter=None, limit=10, at_time=None)` ¶

Return an iterator over collection definitions.

Parameters:

Name	Type	Description	Default
`name_filter`	`Optional[str]`	A JSONPath comparison expression applied to the collection `name` field. The expression is wrapped as `$.name ? (@ <name_filter>)` before being sent to the server. Examples:: `'== "My Ocean Survey"' 'starts with "CTD"' 'like_regex ".ocean." flag "i"'`	`None`
`limit`	`Optional[int]`	Maximum number of collections to return. Default: 10.	`10`
`at_time`	`Optional[datetime]`	Return collections valid at this point in time.	`None`

Returns:

Type	Description
`Iterator[Collection]`	Iterator[Collection]: An iterator over collections.

Source code in ivcap_client/ivcap.py

def list_collections(
    self,
    *,
    name_filter: str | None = None,
    limit: int | None = 10,
    at_time: datetime | None = None,
) -> Iterator[Collection]:
    """Return an iterator over collection definitions.

    Args:
        name_filter (Optional[str]): A JSONPath comparison expression
            applied to the collection ``name`` field.  The expression is
            wrapped as ``$.name ? (@ <name_filter>)`` before being sent
            to the server.

            Examples::

                '== "My Ocean Survey"'
                'starts with "CTD"'
                'like_regex ".*ocean.*" flag "i"'

        limit (Optional[int]): Maximum number of collections to return.
            Default: 10.
        at_time (Optional[datetime]): Return collections valid at this
            point in time.

    Returns:
        Iterator[Collection]: An iterator over collections.
    """
    return list_collections(
        self, name_filter=name_filter, limit=limit, at_time=at_time
    )

`add_to_collection(collection_urn, item_urn, *, policy=None)` ¶

Add an item to a collection with automatic deduplication.

Checks whether item_urn is already a member before creating the membership aspect. If it is already present, returns None (skip silently).

Parameters:

Name	Type	Description	Default
`collection_urn`	`str`	The collection entity URN.	required
`item_urn`	`str`	URN of the entity to add.	required
`policy`	`Optional[URN]`	Optional access policy URN for the membership aspect (`urn:ivcap:policy:…`).	`None`

Returns:

Type	Description
`CollectionItem \| None`	CollectionItem if the item was newly added, `None` if it was already a member.

Source code in ivcap_client/ivcap.py

def add_to_collection(
    self,
    collection_urn: str,
    item_urn: str,
    *,
    policy: URN | None = None,
) -> CollectionItem | None:
    """Add an item to a collection with automatic deduplication.

    Checks whether *item_urn* is already a member before creating the
    membership aspect.  If it is already present, returns ``None``
    (skip silently).

    Args:
        collection_urn (str): The collection entity URN.
        item_urn (str): URN of the entity to add.
        policy (Optional[URN]): Optional access policy URN for the
            membership aspect (``urn:ivcap:policy:…``).

    Returns:
        CollectionItem if the item was newly added, ``None`` if it was
        already a member.
    """
    return add_item_to_collection(self, collection_urn, item_urn, policy=policy)

`remove_from_collection(collection_urn, item_urn)` ¶

Remove an item from a collection by retracting its membership aspect.

Items that are not currently members of the collection are silently skipped.

Parameters:

Name	Type	Description	Default
`collection_urn`	`str`	The collection entity URN.	required
`item_urn`	`str`	URN of the entity to remove.	required

Returns:

Type	Description
`bool`	`True` if the membership aspect was retracted, `False` if the item was not a member.

Source code in ivcap_client/ivcap.py

def remove_from_collection(
    self,
    collection_urn: str,
    item_urn: str,
) -> bool:
    """Remove an item from a collection by retracting its membership aspect.

    Items that are not currently members of the collection are silently
    skipped.

    Args:
        collection_urn (str): The collection entity URN.
        item_urn (str): URN of the entity to remove.

    Returns:
        ``True`` if the membership aspect was retracted, ``False`` if the
        item was not a member.
    """
    return remove_item_from_collection(self, collection_urn, item_urn)

`retract_collection(collection_urn)` ¶

Fully retract a collection and all its item memberships.

All membership aspects are retracted first (paginated), then the collection definition aspect is retracted. This operation cannot be undone.

Parameters:

Name	Type	Description	Default
`collection_urn`	`str`	URN of the collection to retract.	required

Returns:

Type	Description
`int`	Total number of aspect records retracted (items + 1 definition).

Raises:

Type	Description
`ResourceNotFound`	If no collection definition exists for the URN.

Source code in ivcap_client/ivcap.py

def retract_collection(
    self,
    collection_urn: str,
) -> int:
    """Fully retract a collection and all its item memberships.

    All membership aspects are retracted first (paginated), then the
    collection definition aspect is retracted.  This operation cannot
    be undone.

    Args:
        collection_urn (str): URN of the collection to retract.

    Returns:
        Total number of aspect records retracted (items + 1 definition).

    Raises:
        ResourceNotFound: If no collection definition exists for the URN.
    """
    return retract_collection(self, collection_urn)

`list_secrets(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

Return an iterator over all the available secrets fulfilling certain constraints.

Parameters:

Name	Type	Description	Default
`limit`	`Optional[int]`	The 'limit' query option sets the maximum number of items to be included in the result. Default: 10. Example: 10.	`10`
`filter`	`Optional[str]`	The 'filter' system query option allows clients to filter a collection of resources that are addressed by a request URL. The expression specified with 'filter' is evaluated for each resource in the collection, and only items where the expression evaluates to true are included in the response. Example: name ~= 'Scott%'.	`None`

Returns:

Type	Description
`Iterator[Secret]`	Iterator[Secret]: An iterator over a list of secrets

Yields:

Name	Type	Description
`Secret`	`Secret`	A secret object

Source code in ivcap_client/ivcap.py

def list_secrets(
    self,
    *,
    filter: str | None = None,
    limit: int | None = 10,
    order_by: str | None = None,
    order_desc: bool | None = False,
    at_time: datetime.datetime | None = UNSET,
) -> Iterator[Secret]:
    """Return an iterator over all the available secrets fulfilling certain constraints.

    Args:
        limit (Optional[int]): The 'limit' query option sets the maximum number of items
                                to be included in the result. Default: 10. Example: 10.
        filter (Optional[str]): The 'filter' system query option allows clients to filter a
            collection of resources that are addressed by a request URL. The expression specified with 'filter'
                                        is evaluated for each resource in the collection, and only items where the expression
                                        evaluates to true are included in the response. Example: name ~= 'Scott%'.

    Returns:
        Iterator[Secret]: An iterator over a list of secrets

    Yields:
        Secret: A secret object
    """
    kwargs = {
        "filter_": _wrap(filter),
        "limit": _wrap(limit),
        "client": self._client,
    }
    return SecretIter(self, **kwargs)

`search(query)` ¶

Execute query provided in body and return a list of search result.

Parameters:

Name	Type	Description	Default
`query`	`str`	The search query to execute.	required

Raises:

Type	Description
`UnexpectedStatus`	If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
`TimeoutException`	If the request takes longer than Client.timeout.

Returns:

Type	Description
`Any`	Search results.

Source code in ivcap_client/ivcap.py

def search(self, query: str) -> Any:
    """Execute query provided in body and return a list of search result.

    Args:
        query: The search query to execute.

    Raises:
        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
        httpx.TimeoutException: If the request takes longer than Client.timeout.

    Returns:
        Search results.

    """
    body = File(query)
    kwargs = {
        "body": body,
        "content_type": "application/datalog+mangle",
    }
    r = search_search.sync_detailed(client=self._client, **kwargs)
    if r.status_code >= 300:
        return process_error("search", r)
    return r.parsed

`local(base_dir=None)` `classmethod` ¶

Return a :class:LocalIVCAP instance for filesystem-only (no-network) development and testing.

This is the preferred way to force local mode regardless of which environment variables are set — for example in unit tests:

.. code-block:: python

from ivcap_client import IVCAP

ivcap = IVCAP.local(base_dir="./my-artifacts")
artifact = ivcap.upload_artifact(name="result.csv", file_path="/tmp/result.csv")

The base_dir can also be provided via the IVCAP_LOCAL_DIR environment variable. Precedence (highest first):

The base_dir argument to this method.
The IVCAP_LOCAL_DIR environment variable.
The default "ivcap-artifacts".

Parameters:

Name	Type	Description	Default
`base_dir`	`str \| None`	Root directory for artifact and aspect storage. Created on demand. Defaults to `IVCAP_LOCAL_DIR` env var, or `"ivcap-artifacts"` if neither is set.	`None`

Returns:

Name	Type	Description
`LocalIVCAP`	`LocalIVCAP`	A filesystem-backed client with the same `upload_artifact` / `get_artifact` / `add_aspect` interface as :class:`IVCAP`.

Source code in ivcap_client/ivcap.py

@classmethod
def local(cls, base_dir: str | None = None) -> LocalIVCAP:
    """Return a :class:`LocalIVCAP` instance for filesystem-only
    (no-network) development and testing.

    This is the preferred way to force local mode regardless of which
    environment variables are set — for example in unit tests:

    .. code-block:: python

        from ivcap_client import IVCAP

        ivcap = IVCAP.local(base_dir="./my-artifacts")
        artifact = ivcap.upload_artifact(name="result.csv", file_path="/tmp/result.csv")

    The ``base_dir`` can also be provided via the ``IVCAP_LOCAL_DIR``
    environment variable.  Precedence (highest first):

    1. The ``base_dir`` argument to this method.
    2. The ``IVCAP_LOCAL_DIR`` environment variable.
    3. The default ``"ivcap-artifacts"``.

    Args:
        base_dir: Root directory for artifact and aspect storage.
            Created on demand.  Defaults to ``IVCAP_LOCAL_DIR`` env var,
            or ``"ivcap-artifacts"`` if neither is set.

    Returns:
        LocalIVCAP: A filesystem-backed client with the same
        ``upload_artifact`` / ``get_artifact`` / ``add_aspect`` interface
        as :class:`IVCAP`.
    """
    if base_dir is None:
        base_dir = os.environ.get("IVCAP_LOCAL_DIR", "ivcap-artifacts")
    return LocalIVCAP(base_dir=base_dir)

IVCAP Client¶

Quick Reference¶

Class Documentation¶

IVCAP ¶

url property ¶

list_services(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET) ¶

get_service_by_name(name) ¶

get_service(service_id) ¶

get_agent(agent_id) ¶

list_orders(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET) ¶

get_order(order_id) ¶

add_aspect(entity, aspect, *, schema=None, policy=None) ¶

update_aspect(entity, aspect, *, schema=None, policy=None) ¶

list_aspects(*, entity=None, schema=None, content_path=None, at_time=None, limit=10, filter=None, order_by='valid_from', order_direction='DESC', include_content=True) ¶

list_artifacts(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET) ¶

upload_artifact(*, name=None, file_path=None, io_stream=None, content_type=None, content_size=-1, collection=None, policy=None, chunk_size=MAXSIZE, retries=0, retry_delay=30, force_upload=False) ¶

artifact_for_file(file_path) ¶

get_artifact(id) ¶

create_collection(urn, name, *, description=None, policy=None) ¶

get_collection(urn, *, at_time=None) ¶

list_collections(*, name_filter=None, limit=10, at_time=None) ¶

add_to_collection(collection_urn, item_urn, *, policy=None) ¶

remove_from_collection(collection_urn, item_urn) ¶

retract_collection(collection_urn) ¶

list_secrets(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET) ¶

search(query) ¶

local(base_dir=None) classmethod ¶

`IVCAP` ¶

`url` `property` ¶

`list_services(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

`get_service_by_name(name)` ¶

`get_service(service_id)` ¶

`get_agent(agent_id)` ¶

`list_orders(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

`get_order(order_id)` ¶

`add_aspect(entity, aspect, *, schema=None, policy=None)` ¶

`update_aspect(entity, aspect, *, schema=None, policy=None)` ¶

`list_aspects(*, entity=None, schema=None, content_path=None, at_time=None, limit=10, filter=None, order_by='valid_from', order_direction='DESC', include_content=True)` ¶

`list_artifacts(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

`upload_artifact(*, name=None, file_path=None, io_stream=None, content_type=None, content_size=-1, collection=None, policy=None, chunk_size=MAXSIZE, retries=0, retry_delay=30, force_upload=False)` ¶

`artifact_for_file(file_path)` ¶

`get_artifact(id)` ¶

`create_collection(urn, name, *, description=None, policy=None)` ¶

`get_collection(urn, *, at_time=None)` ¶

`list_collections(*, name_filter=None, limit=10, at_time=None)` ¶

`add_to_collection(collection_urn, item_urn, *, policy=None)` ¶

`remove_from_collection(collection_urn, item_urn)` ¶

`retract_collection(collection_urn)` ¶

`list_secrets(*, filter=None, limit=10, order_by=None, order_desc=False, at_time=UNSET)` ¶

`search(query)` ¶

`local(base_dir=None)` `classmethod` ¶