Collection

A Collection is a named, described grouping of IVCAP entity URNs (most commonly artifacts) backed entirely by DataFabric aspects — there is no separate REST endpoint.

Quick Reference

from ivcap_client import IVCAP

ivcap = IVCAP()

# Create (or update) a collection
coll = ivcap.create_collection(
    "urn:ivcap:collection:my-survey",
    name="Ocean CTD Survey",
    description="CTD casts from voyage V2025-03",
)

# Add an item (deduplication is automatic)
item = coll.add_item("urn:ivcap:artifact:<uuid>")

# List items
for ci in coll.items(limit=50):
    print(ci.item)

# Remove a single item
coll.remove_item("urn:ivcap:artifact:<uuid>")

# Retract the entire collection
total = coll.retract()
print(f"Retracted {total} aspect records")

Class Documentation

Collection dataclass

A named, described grouping of entities stored in the IVCAP DataFabric.

A collection is identified by its entity URN and backed by a urn:ivcap:schema:collection.1 aspect. Items are tracked as individual urn:ivcap:schema:collection-item.1 aspects attached to the same entity URN.

Attributes:

Name	Type	Description
urn	str	The collection entity URN (also accessible as :attr:id).
name	str	Human-readable name.
description	str | None	Optional description.
asserter	URN | None	URN of the account that created/updated the definition.
valid_from	datetime | None	Timestamp from which the current definition is valid.
valid_to	datetime | None	Timestamp until which the current definition is valid (None if still active).

Source code in ivcap_client/collection.py

@dataclass
class Collection:
    """A named, described grouping of entities stored in the IVCAP DataFabric.

    A collection is identified by its **entity URN** and backed by a
    ``urn:ivcap:schema:collection.1`` aspect.  Items are tracked as individual
    ``urn:ivcap:schema:collection-item.1`` aspects attached to the same entity
    URN.

    Attributes:
        urn:         The collection entity URN (also accessible as :attr:`id`).
        name:        Human-readable name.
        description: Optional description.
        asserter:    URN of the account that created/updated the definition.
        valid_from:  Timestamp from which the current definition is valid.
        valid_to:    Timestamp until which the current definition is valid
                     (``None`` if still active).
    """

    urn: str
    name: str
    description: str | None = None
    asserter: URN | None = None
    valid_from: datetime.datetime | None = None
    valid_to: datetime.datetime | None = None

    # NOTE: @dataclass with an explicit __init__ keeps the custom __init__.
    def __init__(self, ivcap: IVCAP, **kwargs):
        if not ivcap:
            raise ValueError("missing 'ivcap' argument")
        self._ivcap = ivcap
        self._update(**kwargs)

    def _update(self, **kwargs) -> None:
        self.urn = kwargs.get("urn") or kwargs.get("entity", "")
        self.name = kwargs.get("name", "")
        self.description = kwargs.get("description")
        self.asserter = kwargs.get("asserter")

        vf = kwargs.get("valid_from")
        if isinstance(vf, str):
            vf = datetime.datetime.fromisoformat(vf)
        self.valid_from = vf if not isinstance(vf, Unset) and vf is not None else None

        vt = kwargs.get("valid_to")
        if isinstance(vt, str):
            vt = datetime.datetime.fromisoformat(vt)
        self.valid_to = vt if not isinstance(vt, Unset) and vt is not None else None

    # ------------------------------------------------------------------
    # Properties
    # ------------------------------------------------------------------

    @property
    def id(self) -> str:
        """Alias for :attr:`urn`."""
        return self.urn

    # ------------------------------------------------------------------
    # Methods
    # ------------------------------------------------------------------

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

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

        Args:
            item_urn: URN of the entity to add.
            policy:   Optional access policy URN for the membership aspect
                      (``urn:ivcap:policy:…``).

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

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

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

        Args:
            item_urn: URN of the entity to remove.

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

    def items(
        self,
        *,
        limit: int | None = 10,
        at_time: datetime.datetime | None = None,
    ) -> Iterator[CollectionItem]:
        """Return an iterator over all items in this collection.

        Args:
            limit:   Maximum number of items to yield.  Defaults to 10.
            at_time: Return membership records that were valid at this point in
                     time (ISO 8601).  Defaults to *now*.

        Returns:
            An iterator over :class:`CollectionItem` records.
        """
        kwargs = {
            "entity": self.urn,
            "schema": COLLECTION_ITEM_SCHEMA,
            "include_content": True,
            "limit": _wrap(limit),
            "at_time": _wrap(at_time),
            "client": self._ivcap._client,
        }
        return CollectionItemIter(self._ivcap, **kwargs)

    def retract(self) -> int:
        """Fully retract this 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.

        Returns:
            Total number of aspect records retracted (items + 1 definition).
        """
        return retract_collection(self._ivcap, self.urn)

    def refresh(self) -> Collection:
        """Reload the collection definition from IVCAP.

        Returns:
            ``self`` (updated in-place).
        """
        updated = get_collection(self._ivcap, self.urn)
        self._update(
            urn=updated.urn,
            name=updated.name,
            description=updated.description,
            asserter=updated.asserter,
            valid_from=updated.valid_from,
            valid_to=updated.valid_to,
        )
        return self

    def __repr__(self) -> str:
        return f"<Collection urn={self.urn!r}, name={self.name!r}>"

id property

Alias for :attr:urn.

add_item(item_urn, *, policy=None)

Add an item to this collection with automatic deduplication.

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

Parameters:

Name	Type	Description	Default
item_urn	str	URN of the entity to add.	required
policy	URN | None	Optional access policy URN for the membership aspect (urn:ivcap:policy:…).	None

Returns:

Name	Type	Description
A	CollectionItem | None	class:CollectionItem if the item was newly added, None if it was already a member.

Source code in ivcap_client/collection.py

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

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

    Args:
        item_urn: URN of the entity to add.
        policy:   Optional access policy URN for the membership aspect
                  (``urn:ivcap:policy:…``).

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

remove_item(item_urn)

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

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

Parameters:

Name	Type	Description	Default
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 of the collection.

Source code in ivcap_client/collection.py

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

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

    Args:
        item_urn: URN of the entity to remove.

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

items(*, limit=10, at_time=None)

Return an iterator over all items in this collection.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum number of items to yield. Defaults to 10.	10
at_time	datetime | None	Return membership records that were valid at this point in time (ISO 8601). Defaults to now.	None

Returns:

Type	Description
Iterator[CollectionItem]	An iterator over :class:CollectionItem records.

Source code in ivcap_client/collection.py

def items(
    self,
    *,
    limit: int | None = 10,
    at_time: datetime.datetime | None = None,
) -> Iterator[CollectionItem]:
    """Return an iterator over all items in this collection.

    Args:
        limit:   Maximum number of items to yield.  Defaults to 10.
        at_time: Return membership records that were valid at this point in
                 time (ISO 8601).  Defaults to *now*.

    Returns:
        An iterator over :class:`CollectionItem` records.
    """
    kwargs = {
        "entity": self.urn,
        "schema": COLLECTION_ITEM_SCHEMA,
        "include_content": True,
        "limit": _wrap(limit),
        "at_time": _wrap(at_time),
        "client": self._ivcap._client,
    }
    return CollectionItemIter(self._ivcap, **kwargs)

retract()

Fully retract this 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.

Returns:

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

Source code in ivcap_client/collection.py

def retract(self) -> int:
    """Fully retract this 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.

    Returns:
        Total number of aspect records retracted (items + 1 definition).
    """
    return retract_collection(self._ivcap, self.urn)

refresh()

Reload the collection definition from IVCAP.

Returns:

Type	Description
Collection	self (updated in-place).

Source code in ivcap_client/collection.py

def refresh(self) -> Collection:
    """Reload the collection definition from IVCAP.

    Returns:
        ``self`` (updated in-place).
    """
    updated = get_collection(self._ivcap, self.urn)
    self._update(
        urn=updated.urn,
        name=updated.name,
        description=updated.description,
        asserter=updated.asserter,
        valid_from=updated.valid_from,
        valid_to=updated.valid_to,
    )
    return self

CollectionItem

CollectionItem dataclass

A single membership record linking one item to a collection.

This corresponds to a urn:ivcap:schema:collection-item.1 aspect stored in the DataFabric.

Attributes:

Name	Type	Description
id	str	Aspect record URN.
collection	str	Collection entity URN.
item	str	Member entity URN.
valid_from	datetime | None	Timestamp from which this membership is valid.
valid_to	datetime | None	Timestamp until which this membership is valid (None if still active).

Source code in ivcap_client/collection.py

@dataclass
class CollectionItem:
    """A single membership record linking one item to a collection.

    This corresponds to a ``urn:ivcap:schema:collection-item.1`` aspect stored
    in the DataFabric.

    Attributes:
        id:         Aspect record URN.
        collection: Collection entity URN.
        item:       Member entity URN.
        valid_from: Timestamp from which this membership is valid.
        valid_to:   Timestamp until which this membership is valid (``None``
                    if still active).
    """

    id: str
    collection: str
    item: str
    valid_from: datetime.datetime | None = None
    valid_to: datetime.datetime | None = None

    def __init__(self, **kwargs):
        self.id = kwargs.get("id", "")
        self.collection = kwargs.get("collection", "")
        self.item = kwargs.get("item", "")
        vf = kwargs.get("valid_from")
        self.valid_from = vf if not isinstance(vf, Unset) else None
        vt = kwargs.get("valid_to")
        self.valid_to = vt if not isinstance(vt, Unset) else None

    @property
    def urn(self) -> str:
        """Alias for :attr:`item` — the member entity URN."""
        return self.item

    def __repr__(self) -> str:
        return f"<CollectionItem item={self.item!r}, collection={self.collection!r}>"

urn property

Alias for :attr:item — the member entity URN.

Module-Level Functions

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

Create or update a collection definition (idempotent).

Checks whether a collection.1 aspect already exists for urn:

• If not found — issues a POST (create) to add the definition.
• If found — issues a PUT (update) to replace the definition.

This avoids the server-side "not authorised for method 'add'" error that occurs when a plain PUT is sent for a collection that does not yet exist.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
urn	str	The collection entity URN (must be a valid IVCAP URN, e.g. urn:ivcap:collection:<uuid>).	required
name	str	Human-readable collection name.	required
description	str | None	Optional description.	None
policy	URN | None	Optional access policy URN (urn:ivcap:policy:…).	None

Returns:

Name	Type	Description
A	Collection	class:Collection representing the created/updated definition.

Raises:

Type	Description
	exc:~ivcap_client.exception.MissingParameterValue: If urn or name are empty.

Source code in ivcap_client/collection.py

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

    Checks whether a ``collection.1`` aspect already exists for *urn*:

    * If **not found** — issues a ``POST`` (create) to add the definition.
    * If **found** — issues a ``PUT`` (update) to replace the definition.

    This avoids the server-side "not authorised for method 'add'" error that
    occurs when a plain PUT is sent for a collection that does not yet exist.

    Args:
        ivcap:       The IVCAP client instance.
        urn:         The collection entity URN (must be a valid IVCAP URN,
                     e.g. ``urn:ivcap:collection:<uuid>``).
        name:        Human-readable collection name.
        description: Optional description.
        policy:      Optional access policy URN
                     (``urn:ivcap:policy:…``).

    Returns:
        A :class:`Collection` representing the created/updated definition.

    Raises:
        :exc:`~ivcap_client.exception.MissingParameterValue`:
            If *urn* or *name* are empty.
    """
    if not urn:
        raise MissingParameterValue("Missing collection URN")
    if not name:
        raise MissingParameterValue("Missing collection name")

    body: dict = {"name": name}
    if description:
        body["description"] = description

    # Check whether a collection definition already exists for this URN.
    # Use POST (is_update=False) for new collections and PUT (is_update=True)
    # for updates.  This avoids the server-side "not authorised for method
    # 'add'" error that some deployments raise when a PUT is sent for a
    # collection entity that does not yet have any aspects.
    r = aspect_list.sync_detailed(
        entity=urn,
        schema=COLLECTION_SCHEMA,
        include_content=False,
        limit=1,
        client=ivcap._client,
    )
    if r.status_code >= 300:
        return process_error("create_collection_check", r)

    parsed: AspectListRT = r.parsed
    is_update = bool(parsed.items)

    _add_update_aspect(
        ivcap, is_update, urn, body, schema=COLLECTION_SCHEMA, policy=policy
    )

    return Collection(ivcap, urn=urn, name=name, description=description)

get_collection(ivcap, urn, *, at_time=None)

Fetch a collection definition from IVCAP.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
urn	str	The collection entity URN.	required
at_time	datetime | None	Retrieve the definition as it was at this point in time.	None

Returns:

Name	Type	Description
A	Collection	class:Collection instance.

Raises:

Type	Description
	exc:~ivcap_client.exception.ResourceNotFound: If no collection with the given URN exists.

Source code in ivcap_client/collection.py

def get_collection(
    ivcap: IVCAP,
    urn: str,
    *,
    at_time: datetime.datetime | None = None,
) -> Collection:
    """Fetch a collection definition from IVCAP.

    Args:
        ivcap:   The IVCAP client instance.
        urn:     The collection entity URN.
        at_time: Retrieve the definition as it was at this point in time.

    Returns:
        A :class:`Collection` instance.

    Raises:
        :exc:`~ivcap_client.exception.ResourceNotFound`:
            If no collection with the given URN exists.
    """
    r = aspect_list.sync_detailed(
        entity=urn,
        schema=COLLECTION_SCHEMA,
        include_content=True,
        limit=2,
        at_time=_wrap(at_time),
        client=ivcap._client,
    )
    if r.status_code >= 300:
        return process_error("get_collection", r)

    parsed: AspectListRT = r.parsed
    if not parsed.items:
        raise ResourceNotFound(urn)

    el = parsed.items[0]
    content_dict: dict = {}
    if not isinstance(el.content, Unset):
        content_dict = el.content.to_dict()

    return Collection(
        ivcap,
        urn=el.entity,
        name=content_dict.get("name", ""),
        description=content_dict.get("description"),
        asserter=el.additional_properties.get("asserter"),
        valid_from=el.valid_from,
        valid_to=None if isinstance(el.valid_to, Unset) else el.valid_to,
    )

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

Return an iterator over collection definitions.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
name_filter	str | None	Optional 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	int | None	Maximum number of collections to return. Defaults to 10.	10
at_time	datetime | None	Return collections that were valid at this point in time.	None

Returns:

Type	Description
Iterator[Collection]	An iterator over :class:Collection objects.

Source code in ivcap_client/collection.py

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

    Args:
        ivcap:       The IVCAP client instance.
        name_filter: Optional 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:       Maximum number of collections to return.  Defaults to 10.
        at_time:     Return collections that were valid at this point in time.

    Returns:
        An iterator over :class:`Collection` objects.
    """
    content_path: str | None = None
    if name_filter:
        content_path = f"$.name ? (@ {name_filter})"

    kwargs = {
        "schema": COLLECTION_SCHEMA,
        "include_content": True,
        "limit": _wrap(limit),
        "at_time": _wrap(at_time),
        "content_path": _wrap(content_path),
        "client": ivcap._client,
    }
    return CollectionIter(ivcap, **kwargs)

add_item_to_collection(ivcap, collection_urn, item_urn, *, policy=None)

Add an item to a collection with a prior deduplication check.

Before creating the membership aspect the server is queried to confirm that no active collection-item.1 aspect already records this (collection, item) pair. If found the call returns None (skip silently). Otherwise a new aspect record is created with POST.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
collection_urn	str	The collection entity URN.	required
item_urn	str	URN of the entity to add as a member.	required
policy	URN | None	Optional access policy URN for the membership aspect (urn:ivcap:policy:…).	None

Returns:

Name	Type	Description
A	CollectionItem | None	class:CollectionItem if the item was added, None if it was already a member.

Raises:

Type	Description
	exc:~ivcap_client.exception.MissingParameterValue: If collection_urn or item_urn are empty.

Source code in ivcap_client/collection.py

def add_item_to_collection(
    ivcap: IVCAP,
    collection_urn: str,
    item_urn: str,
    *,
    policy: URN | None = None,
) -> CollectionItem | None:
    """Add an item to a collection with a prior deduplication check.

    Before creating the membership aspect the server is queried to confirm
    that no active ``collection-item.1`` aspect already records this
    ``(collection, item)`` pair.  If found the call returns ``None`` (skip
    silently).  Otherwise a new aspect record is created with ``POST``.

    Args:
        ivcap:          The IVCAP client instance.
        collection_urn: The collection entity URN.
        item_urn:       URN of the entity to add as a member.
        policy:         Optional access policy URN for the membership aspect
                        (``urn:ivcap:policy:…``).

    Returns:
        A :class:`CollectionItem` if the item was added, ``None`` if it
        was already a member.

    Raises:
        :exc:`~ivcap_client.exception.MissingParameterValue`:
            If *collection_urn* or *item_urn* are empty.
    """
    if not collection_urn:
        raise MissingParameterValue("Missing collection URN")
    if not item_urn:
        raise MissingParameterValue("Missing item URN")

    # Dedup check — use JSONPath filter to avoid fetching full content.
    # Some deployments do not support JSONPath content_path filter expressions
    # and return 5xx/EOF for such queries.  In that case we fall through to
    # the POST rather than failing hard; the worst outcome is a duplicate
    # membership aspect, which is benign.
    dedup_path = f'$.item ? (@ == "{item_urn}")'
    r = aspect_list.sync_detailed(
        entity=collection_urn,
        schema=COLLECTION_ITEM_SCHEMA,
        content_path=dedup_path,
        include_content=False,
        limit=1,
        client=ivcap._client,
    )
    if r.status_code < 300:
        parsed: AspectListRT = r.parsed
        if parsed.items:
            # Already a member — skip silently.
            return None
    # If the dedup check returned 4xx/5xx, fall through to POST.

    # Create the membership aspect (POST — append, not replace).
    body = {
        "collection": collection_urn,
        "item": item_urn,
    }
    aspect = _add_update_aspect(
        ivcap, False, collection_urn, body, schema=COLLECTION_ITEM_SCHEMA, policy=policy
    )
    return CollectionItem(
        id=aspect.id,
        collection=collection_urn,
        item=item_urn,
    )

remove_item_from_collection(ivcap, collection_urn, item_urn)

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

Looks up the active collection-item.1 aspect for the (collection_urn, item_urn) pair using a JSONPath filter, then retracts it via DELETE /1/aspects/<id>. Items that are not currently members of the collection are silently skipped.

This operation leaves a retracted (inactive) aspect record in the DataFabric — the store is append-only by design. The item can be re-added later by calling :func:add_item_to_collection again.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
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 found and retracted, False if the item was not a member.

Raises:

Type	Description
	exc:~ivcap_client.exception.MissingParameterValue: If collection_urn or item_urn are empty.

Source code in ivcap_client/collection.py

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

    Looks up the active ``collection-item.1`` aspect for the
    ``(collection_urn, item_urn)`` pair using a JSONPath filter, then
    retracts it via ``DELETE /1/aspects/<id>``.  Items that are not
    currently members of the collection are silently skipped.

    This operation leaves a retracted (inactive) aspect record in the
    DataFabric — the store is append-only by design.  The item can be
    re-added later by calling :func:`add_item_to_collection` again.

    Args:
        ivcap:          The IVCAP client instance.
        collection_urn: The collection entity URN.
        item_urn:       URN of the entity to remove.

    Returns:
        ``True`` if the membership aspect was found and retracted,
        ``False`` if the item was not a member.

    Raises:
        :exc:`~ivcap_client.exception.MissingParameterValue`:
            If *collection_urn* or *item_urn* are empty.
    """
    if not collection_urn:
        raise MissingParameterValue("Missing collection URN")
    if not item_urn:
        raise MissingParameterValue("Missing item URN")

    # Find the active membership aspect for this item.
    find_path = f'$.item ? (@ == "{item_urn}")'
    r = aspect_list.sync_detailed(
        entity=collection_urn,
        schema=COLLECTION_ITEM_SCHEMA,
        content_path=find_path,
        include_content=False,
        limit=1,
        client=ivcap._client,
    )
    if r.status_code >= 300:
        return process_error("remove_item_from_collection_find", r)

    parsed: AspectListRT = r.parsed
    if not parsed.items:
        # Not a member — skip silently.
        return False

    aspect_id = parsed.items[0].id
    r2 = aspect_retract.sync_detailed(aspect_id, client=ivcap._client)
    if r2.status_code >= 300:
        return process_error("remove_item_from_collection_retract", r2)

    return True

retract_collection(ivcap, collection_urn)

Fully retract a collection and all its item memberships.

All active collection-item.1 aspects for collection_urn are retracted first (one DELETE per membership record, paginated), then the collection.1 definition aspect is retracted.

The DataFabric is append-only — this leaves retracted (inactive) records behind but makes the collection invisible to normal queries. The operation cannot be undone.

Parameters:

Name	Type	Description	Default
ivcap	IVCAP	The IVCAP client instance.	required
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
	exc:~ivcap_client.exception.MissingParameterValue: If collection_urn is empty.
	exc:~ivcap_client.exception.ResourceNotFound: If no collection definition aspect exists for collection_urn.

Source code in ivcap_client/collection.py

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

    All active ``collection-item.1`` aspects for *collection_urn* are
    retracted first (one ``DELETE`` per membership record, paginated), then
    the ``collection.1`` definition aspect is retracted.

    The DataFabric is append-only — this leaves retracted (inactive) records
    behind but makes the collection invisible to normal queries.  The
    operation cannot be undone.

    Args:
        ivcap:          The IVCAP client instance.
        collection_urn: URN of the collection to retract.

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

    Raises:
        :exc:`~ivcap_client.exception.MissingParameterValue`:
            If *collection_urn* is empty.
        :exc:`~ivcap_client.exception.ResourceNotFound`:
            If no collection definition aspect exists for *collection_urn*.
    """
    if not collection_urn:
        raise MissingParameterValue("Missing collection URN")

    retracted = 0

    # ------------------------------------------------------------------
    # Step 1 — retract all collection-item.1 membership aspects
    # ------------------------------------------------------------------
    page_cursor = UNSET
    while True:
        kwargs: dict = {
            "entity": collection_urn,
            "schema": COLLECTION_ITEM_SCHEMA,
            "include_content": False,
            "limit": 50,
            "client": ivcap._client,
        }
        if not isinstance(page_cursor, Unset):
            kwargs["page"] = page_cursor

        r = aspect_list.sync_detailed(**kwargs)
        if r.status_code >= 300:
            return process_error("retract_collection_list_items", r)

        parsed: AspectListRT = r.parsed
        for item in parsed.items:
            r2 = aspect_retract.sync_detailed(item.id, client=ivcap._client)
            if r2.status_code >= 300:
                return process_error("retract_collection_retract_item", r2)
            retracted += 1

        # Advance the pagination cursor.
        links = Links(parsed.links)
        if links.next is None:
            break
        page_cursor = links.next

    # ------------------------------------------------------------------
    # Step 2 — retract the collection definition aspect
    # ------------------------------------------------------------------
    r = aspect_list.sync_detailed(
        entity=collection_urn,
        schema=COLLECTION_SCHEMA,
        include_content=False,
        limit=1,
        client=ivcap._client,
    )
    if r.status_code >= 300:
        return process_error("retract_collection_find_def", r)

    parsed = r.parsed
    if not parsed.items:
        raise ResourceNotFound(collection_urn)

    r2 = aspect_retract.sync_detailed(parsed.items[0].id, client=ivcap._client)
    if r2.status_code >= 300:
        return process_error("retract_collection_retract_def", r2)
    retracted += 1

    return retracted

Schema Constants

Constant	Value
COLLECTION_SCHEMA	urn:ivcap:schema:collection.1
COLLECTION_ITEM_SCHEMA	urn:ivcap:schema:collection-item.1

See Also

• Guide: Collections — Usage patterns and examples
• API Reference: IVCAP Client — create_collection(), get_collection(), list_collections(), add_to_collection(), remove_from_collection(), retract_collection()
• API Reference: Aspect — The underlying aspect primitives

Copy Markdown