DomoFileset

DomoFileset

DomoFileset dataclass

DomoFileset(
    auth: DomoAuth,
    id: str,
    raw: dict = dict(),
    name: str | None = None,
    description: str | None = None,
    ai_enabled: bool | None = None,
    index_status: str | None = None,
    connector: str | None = None,
    created: str | None = None,
    created_by: int | None = None,
    updated: str | None = None,
    updated_by: int | None = None,
    owner: str | None = None,
    size: int | None = None,
    file_count: int | None = None,
    account_id: int | None = None,
    connector_context: str | None = None,
    permission: str | None = None,
)

Bases: DomoEntity

Represents a single Domo Fileset entity.

display_url property

display_url: str

Return the Domo UI URL for this fileset.

delete async

delete(
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bool | ResponseGetData

Delete this fileset.

Parameters:

Name	Type	Description	Default
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
bool | ResponseGetData	True when the delete succeeded, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def delete(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bool | ResponseGetData:
    """Delete this fileset.

    Args:
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        ``True`` when the delete succeeded, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.delete_fileset(
        fileset_id=self.id,
        auth=self.auth,
        context=context,
    )

    if return_raw:
        return res

    return res.is_success

from_dict classmethod

from_dict(
    auth: DomoAuth, obj: dict[str, Any]
) -> DomoFileset

Build a fileset from an API payload.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
obj	dict[str, Any]	Fileset payload as returned by the fileset routes. Missing keys become None.	required

Returns:

Type	Description
DomoFileset	The populated DomoFileset.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]) -> DomoFileset:
    """Build a fileset from an API payload.

    Args:
        auth: Authentication object for Domo APIs.
        obj: Fileset payload as returned by the fileset routes. Missing
            keys become ``None``.

    Returns:
        The populated ``DomoFileset``.
    """
    return cls(
        auth=auth,
        id=str(obj.get("id")),
        name=obj.get("name"),
        description=obj.get("description"),
        ai_enabled=obj.get("aiEnabled"),
        index_status=obj.get("indexStatus"),
        connector=obj.get("connector"),
        created=obj.get("created"),
        created_by=obj.get("createdBy"),
        updated=obj.get("updated"),
        updated_by=obj.get("updatedBy"),
        owner=obj.get("owner"),
        size=obj.get("size"),
        file_count=obj.get("fileCount"),
        account_id=obj.get("accountId"),
        connector_context=obj.get("connectorContext"),
        permission=obj.get("permission"),
        raw=obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    fileset_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoFileset | ResponseGetData

Retrieve a fileset by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
fileset_id	str	The fileset ID.	required
debug_num_stacks_to_drop	int	Stack frames to drop in debug output.	2
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
DomoFileset | ResponseGetData	A DomoFileset, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    fileset_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoFileset | ResponseGetData:
    """Retrieve a fileset by ID.

    Args:
        auth: Authentication object for Domo APIs.
        fileset_id: The fileset ID.
        debug_num_stacks_to_drop: Stack frames to drop in debug output.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        A ``DomoFileset``, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await fileset_routes.get_fileset_by_id(
        auth=auth,
        fileset_id=fileset_id,
        context=context,
    )

    if return_raw:
        return res

    return cls.from_dict(auth=auth, obj=res.response)

get_entity_by_id async classmethod

get_entity_by_id(
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs: Any
) -> DomoFileset

Internal wrapper used by the base refresh() method.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
entity_id	str	The fileset ID.	required
debug_num_stacks_to_drop	int	Stack frames to drop in debug output.	2
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
check_if_published	bool | None	Unused; accepted for base-class compatibility.	None
parent_auth_retrieval_fn	Any | None	Unused; accepted for base-class compatibility.	None
parent_auth	Any | None	Unused; accepted for base-class compatibility.	None
**kwargs	Any	Ignored additional parameters.	{}

Returns:

Type	Description
DomoFileset	The retrieved DomoFileset.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

@classmethod
async def get_entity_by_id(
    cls,
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs: Any,
) -> DomoFileset:
    """Internal wrapper used by the base ``refresh()`` method.

    Args:
        auth: Authentication object for Domo APIs.
        entity_id: The fileset ID.
        debug_num_stacks_to_drop: Stack frames to drop in debug output.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        check_if_published: Unused; accepted for base-class compatibility.
        parent_auth_retrieval_fn: Unused; accepted for base-class compatibility.
        parent_auth: Unused; accepted for base-class compatibility.
        **kwargs: Ignored additional parameters.

    Returns:
        The retrieved ``DomoFileset``.
    """
    return await cls.get_by_id(
        auth=auth,
        fileset_id=entity_id,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        debug_api=debug_api,
        session=session,
    )

update async

update(
    body: dict[str, Any],
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoFileset | ResponseGetData

Update this fileset, then refresh it.

Parameters:

Name	Type	Description	Default
body	dict[str, Any]	Fields to update.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
DomoFileset | ResponseGetData	This fileset (refreshed), or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def update(
    self,
    body: dict[str, Any],
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoFileset | ResponseGetData:
    """Update this fileset, then refresh it.

    Args:
        body: Fields to update.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        This fileset (refreshed), or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.update_fileset(
        fileset_id=self.id,
        auth=self.auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    await self.refresh()
    return self

DomoFilesetAccessController

DomoFilesetAccessController(auth: DomoAuth, parent)

Bases: DomoAccessRelationshipController

Access controller for Domo filesets.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

def __init__(self, auth: DomoAuth, parent):
    super().__init__(auth=auth, parent_object=parent)
    self.parent = parent
    self.relationships: list[Relationship] = []

add_owners async

add_owners(
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Deprecated alias for transfer_owner — which TRANSFERS, not adds.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def add_owners(
    self,
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Deprecated alias for `transfer_owner` — which TRANSFERS, not adds."""
    warnings.warn(
        "DomoFilesetAccessController.add_owners is deprecated; use "
        "transfer_owner instead. It TRANSFERS ownership — the current owner "
        "is displaced and there is no undo (remove_owners raises "
        "NotImplementedError).",
        DeprecationWarning,
        stacklevel=2,
    )
    return await self.transfer_owner(
        user_id=user_id, context=context, **context_kwargs
    )

create_relationship async

create_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Create an access relationship for the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an access relationship for the fileset."""
    return await self.share(
        user_ids=(
            [int(target_entity_id)]
            if target_entity_type == EntityType.USER
            else None
        ),
        group_ids=(
            [int(target_entity_id)]
            if target_entity_type == EntityType.GROUP
            else None
        ),
        **kwargs,
    )

get_accesslist async

get_accesslist(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Relationship]

Get access list for the fileset as relationships.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def get_accesslist(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get access list for the fileset as relationships."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await fileset_routes.get_fileset_access(
        auth=self.auth,
        fileset_id=self.parent_id,
        context=context,
    )

    if return_raw:
        return res

    relationships: list[Relationship] = []
    access_entries: list[dict[str, Any]] = []

    if isinstance(res.response, dict):
        access_entries = res.response.get("fileSetAccess", [])

    for entry in access_entries:
        entity_id = entry.get("entityId")
        entity_type = entry.get("entityType")
        permission = entry.get("permission")
        if entity_id is None or not entity_type:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(entity_id),
                from_entity_type=(
                    EntityType.GROUP
                    if str(entity_type).upper() == "GROUP"
                    else EntityType.USER
                ),
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.DATASET,  # FILESET not in EntityType enum
                relationship_type=self._map_permission(permission),
                metadata=entry,
            )
        )

    self.relationships = relationships
    self.invalidate_cache()
    return relationships

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get direct relationships for the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def get_direct_relationships(self) -> list[Relationship]:
    """Get direct relationships for the fileset."""
    return await self.get_accesslist()

get_owners async

get_owners(
    *, context: RouteContext | None = None, **context_kwargs
) -> list[Relationship]

Get owner relationships for the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def get_owners(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get owner relationships for the fileset."""
    if not self.relationships:
        await self.get_accesslist(context=context, **context_kwargs)

    return [
        rel
        for rel in self.relationships
        if rel.relationship_type == RelationshipType.HAS_ACCESS_OWNER
    ]

get_stats async

get_stats(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> dict | Any

Get file type counts and index status for the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def get_stats(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> dict | Any:
    """Get file type counts and index status for the fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await fileset_routes.get_fileset_stats(
        auth=self.auth,
        fileset_id=self.parent_id,
        context=context,
    )

    if return_raw:
        return res

    return res.response

remove_owners async

remove_owners(**_kwargs) -> bool

Remove owners from the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def remove_owners(
    self,
    **_kwargs,
) -> bool:
    """Remove owners from the fileset."""
    raise NotImplementedError(
        "Removing fileset owners is not supported by the Domo API."
    )

remove_relationship async

remove_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Remove an access relationship for the fileset.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an access relationship for the fileset."""
    raise NotImplementedError(
        "Removing fileset access entries is not supported by the Domo API."
    )

share async

share(
    user_ids: list[int] | None = None,
    group_ids: list[int] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Share the fileset with users or groups (READ access only).

Note: The Domo fileset API only supports 'READ' as a non-owner permission. Owner permission cannot be modified via the access endpoint.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def share(
    self,
    user_ids: list[int] | None = None,
    group_ids: list[int] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Share the fileset with users or groups (READ access only).

    Note: The Domo fileset API only supports 'READ' as a non-owner permission.
    Owner permission cannot be modified via the access endpoint.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    access_entries = []
    for uid in user_ids or []:
        access_entries.append(
            {"entityId": uid, "entityType": "USER", "permission": "READ"}
        )
    for gid in group_ids or []:
        access_entries.append(
            {"entityId": gid, "entityType": "GROUP", "permission": "READ"}
        )

    if not access_entries:
        raise ValueError("Must provide user_ids or group_ids to share fileset")

    body = {"fileSetAccess": access_entries}

    res = await fileset_routes.share_fileset(
        auth=self.auth,
        fileset_id=self.parent_id,
        body=body,
        context=context,
    )

    # Invalidate cache so next access fetches fresh data
    self.invalidate_cache()
    return res.is_success

transfer_owner async

transfer_owner(
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

TRANSFER ownership of the fileset to user_id.

This is not an "add". It calls transfer_fileset_ownership, which moves the fileset's owner — and remove_owners raises NotImplementedError, so there is no undo through this library.

⚠️ The access list will not show you that this happened. Verified on domo-community 2026-08-31 with a throwaway fileset: after the call the fileset's scalar owner had moved from the caller to user_id, but get_fileset_access listed BOTH principals with permission OWNER. Read back get_fileset_by_id(...).response["owner"], not the access list, to see who owns it now — checking the ACL makes a transfer look additive.

Named add_owners until it was renamed; add_owners remains as a deprecated alias. The old name shared a spelling with DomoCardAccessController.add_owners and DomoPageAccessController.add_owners, which really are additive, so the same call read as safe on three controllers and was destructive on one.

See .agents/guides/access-and-sharing.md.

Source code in src/crew_dcs/classes/DomoFileset/access_controller.py

async def transfer_owner(
    self,
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """TRANSFER ownership of the fileset to `user_id`.

    This is not an "add". It calls `transfer_fileset_ownership`, which moves
    the fileset's `owner` — and `remove_owners` raises NotImplementedError,
    so there is no undo through this library.

    ⚠️ **The access list will not show you that this happened.** Verified on
    domo-community 2026-08-31 with a throwaway fileset: after the call the
    fileset's scalar `owner` had moved from the caller to `user_id`, but
    `get_fileset_access` listed BOTH principals with permission OWNER. Read
    back `get_fileset_by_id(...).response["owner"]`, not the access list, to
    see who owns it now — checking the ACL makes a transfer look additive.

    Named `add_owners` until it was renamed; `add_owners` remains as a
    deprecated alias. The old name shared a spelling with
    `DomoCardAccessController.add_owners` and
    `DomoPageAccessController.add_owners`, which really are additive, so the
    same call read as safe on three controllers and was destructive on one.

    See `.agents/guides/access-and-sharing.md`.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await fileset_routes.transfer_fileset_ownership(
        auth=self.auth,
        fileset_id=self.parent_id,
        user_id=user_id,
        context=context,
    )

    # Invalidate cache so next access fetches fresh data
    self.invalidate_cache()
    return res.is_success

DomoFileset_Files dataclass

DomoFileset_Files(parent: DomoEntity)

Bases: DomoSubEntity

Sub-entity for interacting with files within a DomoFileset.

delete_by_id async

delete_by_id(
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bool | ResponseGetData

Delete a file from this fileset by file ID.

Parameters:

Name	Type	Description	Default
file_id	str	The file ID.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
bool | ResponseGetData	True when the delete succeeded, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def delete_by_id(
    self,
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bool | ResponseGetData:
    """Delete a file from this fileset by file ID.

    Args:
        file_id: The file ID.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        ``True`` when the delete succeeded, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.delete_fileset_file(
        auth=self.auth,
        fileset_id=self.parent.id,
        file_id=file_id,
        context=context,
    )

    if return_raw:
        return res

    return res.is_success

delete_by_path async

delete_by_path(
    file_path: str,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bool | ResponseGetData

Delete a file from this fileset by path.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file within the fileset.	required
body	dict[str, Any] | None	Optional custom request body; overrides file_path.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
bool | ResponseGetData	True when the delete succeeded, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def delete_by_path(
    self,
    file_path: str,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bool | ResponseGetData:
    """Delete a file from this fileset by path.

    Args:
        file_path: Path of the file within the fileset.
        body: Optional custom request body; overrides ``file_path``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        ``True`` when the delete succeeded, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not body:
        body = {"filePath": file_path}

    res = await fileset_routes.delete_fileset_file_by_path(
        auth=self.auth,
        fileset_id=self.parent.id,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    return res.is_success

download async

download(
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bytes | ResponseGetData

Download a file from this fileset by file ID.

Parameters:

Name	Type	Description	Default
file_id	str	The file ID.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
bytes | ResponseGetData	The raw file bytes, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def download(
    self,
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bytes | ResponseGetData:
    """Download a file from this fileset by file ID.

    Args:
        file_id: The file ID.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The raw file bytes, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.download_fileset_file(
        auth=self.auth,
        fileset_id=self.parent.id,
        file_id=file_id,
        context=context,
    )

    if return_raw:
        return res

    return res.response

download_by_path async

download_by_path(
    file_path: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bytes | ResponseGetData

Download a file from this fileset by path.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file within the fileset, e.g. dir/name.md.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
bytes | ResponseGetData	The raw file bytes, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def download_by_path(
    self,
    file_path: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bytes | ResponseGetData:
    """Download a file from this fileset by path.

    Args:
        file_path: Path of the file within the fileset, e.g. ``dir/name.md``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The raw file bytes, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.download_fileset_file_by_path(
        auth=self.auth,
        fileset_id=self.parent.id,
        path=file_path,
        context=context,
    )

    if return_raw:
        return res

    return res.response

get_by_id async

get_by_id(
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

Get a file from this fileset by file ID.

Parameters:

Name	Type	Description	Default
file_id	str	The file ID.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	The file payload, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def get_by_id(
    self,
    file_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """Get a file from this fileset by file ID.

    Args:
        file_id: The file ID.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The file payload, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.get_fileset_file(
        auth=self.auth,
        fileset_id=self.parent.id,
        file_id=file_id,
        context=context,
    )

    if return_raw:
        return res

    return res.response

get_by_path async

get_by_path(
    file_path: str,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

Get a file from this fileset by path.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file within the fileset.	required
body	dict[str, Any] | None	Optional custom request body; overrides file_path.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	The file payload, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def get_by_path(
    self,
    file_path: str,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """Get a file from this fileset by path.

    Args:
        file_path: Path of the file within the fileset.
        body: Optional custom request body; overrides ``file_path``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The file payload, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not body:
        body = {"filePath": file_path}

    res = await fileset_routes.get_fileset_file_by_path(
        auth=self.auth,
        fileset_id=self.parent.id,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    return res.response

search async

search(
    directory_path: str | None = None,
    loop_until_end: bool = False,
    limit: int = 100,
    maximum: int = 0,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_loop: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[dict] | ResponseGetData

Search files within this fileset.

Parameters:

Name	Type	Description	Default
directory_path	str | None	Optional directory path to scope the search.	None
loop_until_end	bool	If True, retrieve all files across all pages.	False
limit	int	Maximum files per request (pagination).	100
maximum	int	Maximum total files to retrieve (0 = no limit).	0
body	dict[str, Any] | None	Optional custom search body.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
debug_loop	bool	If True, print pagination debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def search(
    self,
    directory_path: str | None = None,
    loop_until_end: bool = False,
    limit: int = 100,
    maximum: int = 0,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_loop: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[dict] | ResponseGetData:
    """Search files within this fileset.

    Args:
        directory_path: Optional directory path to scope the search.
        loop_until_end: If True, retrieve all files across all pages.
        limit: Maximum files per request (pagination).
        maximum: Maximum total files to retrieve (0 = no limit).
        body: Optional custom search body.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        debug_loop: If True, print pagination debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.search_fileset_files(
        auth=self.auth,
        domo_fileset_id=self.parent.id,
        body=body,
        directory_path=directory_path or "",
        immediate_children=False if loop_until_end else (not directory_path),
        loop_until_end=loop_until_end,
        limit=limit,
        maximum=maximum,
        debug_loop=debug_loop,
        context=context,
    )

    if return_raw:
        return res

    if isinstance(res.response, list):
        return res.response
    if isinstance(res.response, dict) and "files" in res.response:
        return res.response["files"]
    return []

upload async

upload(
    file_name: str,
    file_content: bytes | str,
    directory_path: str | None = None,
    content_type: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

Upload a file to this fileset.

Parameters:

Name	Type	Description	Default
file_name	str	Name of the file, including extension.	required
file_content	bytes | str	File content as bytes or text.	required
directory_path	str | None	Optional target directory within the fileset.	None
content_type	str | None	MIME type; defaults to text/plain.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	The upload response payload, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def upload(
    self,
    file_name: str,
    file_content: bytes | str,
    directory_path: str | None = None,
    content_type: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """Upload a file to this fileset.

    Args:
        file_name: Name of the file, including extension.
        file_content: File content as bytes or text.
        directory_path: Optional target directory within the fileset.
        content_type: MIME type; defaults to ``text/plain``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The upload response payload, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.upload_fileset_file(
        auth=self.auth,
        fileset_id=self.parent.id,
        file_name=file_name,
        file_content=file_content,
        directory_path=directory_path or "",
        content_type=content_type or "text/plain",
        context=context,
    )

    if return_raw:
        return res

    return res.response

DomoFileset_Index dataclass

DomoFileset_Index(parent: DomoEntity)

Bases: DomoSubEntity

Sub-entity for interacting with index/AI features of a DomoFileset.

create async

create(
    index_id: str,
    embedding_model: str = "domo.domo_ai.domo-embed-text-multilingual-v1:cohere",
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

Create a vectorDB index for this fileset.

Parameters:

Name	Type	Description	Default
index_id	str	ID for the new index.	required
embedding_model	str	Embedding model identifier.	'domo.domo_ai.domo-embed-text-multilingual-v1:cohere'
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	The created index payload, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def create(
    self,
    index_id: str,
    embedding_model: str = "domo.domo_ai.domo-embed-text-multilingual-v1:cohere",
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """Create a vectorDB index for this fileset.

    Args:
        index_id: ID for the new index.
        embedding_model: Embedding model identifier.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The created index payload, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.create_filesets_index(
        auth=self.auth,
        index_id=index_id,
        embedding_model=embedding_model,
        context=context,
    )

    if return_raw:
        return res

    return res.response

get_access async

get_access(
    index_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

Get the access list for a Recall (vectorDB) index.

Parameters:

Name	Type	Description	Default
index_id	str	The ID of the Recall index, as passed to create.	required
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	The access list payload, or the raw ResponseGetData when
dict | ResponseGetData	return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def get_access(
    self,
    index_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """Get the access list for a Recall (vectorDB) index.

    Args:
        index_id: The ID of the Recall index, as passed to ``create``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The access list payload, or the raw ``ResponseGetData`` when
        ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.get_recall_index_access(
        auth=self.auth,
        index_id=index_id,
        context=context,
    )

    if return_raw:
        return res

    return res.response

query async

query(
    query_text: str,
    directory_path: str | None = None,
    top_k: int | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[dict] | ResponseGetData

Query the fileset index for relevant documents.

Parameters:

Name	Type	Description	Default
query_text	str	Natural-language query.	required
directory_path	str | None	Optional directory to scope the query.	None
top_k	int | None	Optional maximum number of matches.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
list[dict] | ResponseGetData	The query response payload, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def query(
    self,
    query_text: str,
    directory_path: str | None = None,
    top_k: int | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[dict] | ResponseGetData:
    """Query the fileset index for relevant documents.

    Args:
        query_text: Natural-language query.
        directory_path: Optional directory to scope the query.
        top_k: Optional maximum number of matches.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The query response payload, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.query_fileset(
        auth=self.auth,
        fileset_id=self.parent.id,
        query=query_text,
        directory_path=directory_path,
        top_k=top_k,
        context=context,
    )

    if return_raw:
        return res

    return res.response if isinstance(res.response, list) else res.response  # noqa: RUF034

rag_chat async

rag_chat(
    question: str,
    *,
    directory_path: str | None = None,
    top_k: int = 5,
    score_threshold: float = 0.6,
    llm_url: str | None = None,
    llm_model: str = "qwen3.5-9b",
    llm_api_key: str = "no-key-required",
    system_prompt: str | None = None,
    return_context: bool = False,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict | ResponseGetData

RAG chat: query fileset, assemble context, synthesize answer with LLM.

This is the "Option C" convenience method that wraps: 1. fileset_routes.query_fileset() — retrieve relevant chunks 2. Context assembly — filter by score, format chunks 3. LLM synthesis — call external LLM (gateway) or Domo native LLM

Parameters:

Name	Type	Description	Default
question	str	The user's question.	required
directory_path	str | None	Optional directory to scope the search (e.g., "api-reference/").	None
top_k	int	Maximum number of chunks to retrieve.	5
score_threshold	float	Minimum relevance score to include (default 0.6).	0.6
llm_url	str | None	External LLM endpoint URL (e.g., "http://localhost:7630/v1/chat/completions"). If None, attempts to use Domo native LLM (may not be available on all instances).	None
llm_model	str	Model name for external LLM (default "qwen3.5-9b").	'qwen3.5-9b'
llm_api_key	str	API key for external LLM (default "no-key-required" for gateway).	'no-key-required'
system_prompt	str | None	Override the default RAG system prompt.	None
return_context	bool	If True, include raw context chunks in response.	False
return_raw	bool	If True, return raw ResponseGetData from LLM call.	False
debug_api	bool	Enable API debug logging.	False
session	AsyncClient | None	Optional shared httpx.AsyncClient.	None
context	RouteContext | None	Route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
dict | ResponseGetData	dict with: - "answer": The synthesized answer string. - "sources": List of {"path": str, "score": float}. - "context": List of chunk dicts (only if return_context=True).

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def rag_chat(
    self,
    question: str,
    *,
    directory_path: str | None = None,
    top_k: int = 5,
    score_threshold: float = 0.6,
    llm_url: str | None = None,
    llm_model: str = "qwen3.5-9b",
    llm_api_key: str = "no-key-required",
    system_prompt: str | None = None,
    return_context: bool = False,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict | ResponseGetData:
    """RAG chat: query fileset, assemble context, synthesize answer with LLM.

    This is the "Option C" convenience method that wraps:
    1. fileset_routes.query_fileset() — retrieve relevant chunks
    2. Context assembly — filter by score, format chunks
    3. LLM synthesis — call external LLM (gateway) or Domo native LLM

    Args:
        question: The user's question.
        directory_path: Optional directory to scope the search (e.g., "api-reference/").
        top_k: Maximum number of chunks to retrieve.
        score_threshold: Minimum relevance score to include (default 0.6).
        llm_url: External LLM endpoint URL (e.g., "http://localhost:7630/v1/chat/completions").
                  If None, attempts to use Domo native LLM (may not be available on all instances).
        llm_model: Model name for external LLM (default "qwen3.5-9b").
        llm_api_key: API key for external LLM (default "no-key-required" for gateway).
        system_prompt: Override the default RAG system prompt.
        return_context: If True, include raw context chunks in response.
        return_raw: If True, return raw ResponseGetData from LLM call.
        debug_api: Enable API debug logging.
        session: Optional shared httpx.AsyncClient.
        context: Route context.
        **context_kwargs: Additional context parameters.

    Returns:
        dict with:
            - "answer": The synthesized answer string.
            - "sources": List of {"path": str, "score": float}.
            - "context": List of chunk dicts (only if return_context=True).
    """
    from ...routes import ai as ai_routes

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    # Step 1: Query fileset for relevant chunks
    query_res = await fileset_routes.query_fileset(
        auth=self.auth,
        fileset_id=self.parent.id,
        query=question,
        directory_path=directory_path,
        top_k=top_k,
        context=context,
    )

    matches = query_res.response.get("matches", [])

    # Step 2: Filter by score threshold
    filtered = [m for m in matches if m.get("score", 0) >= score_threshold]

    if not filtered:
        return {
            "answer": "I couldn't find any relevant information in the indexed documents.",
            "sources": [],
            "context": [] if return_context else None,
        }

    # Step 3: Assemble context from chunks
    context_chunks = []
    sources = []
    for m in filtered:
        metadata = m.get("metadata", {})
        content = m.get("content", {})
        path = metadata.get("path", "unknown")
        score = m.get("score", 0)

        text = content.get("text", "") if isinstance(content, dict) else ""
        if not text:
            text = m.get("content", "") if isinstance(m.get("content"), str) else ""

        context_chunks.append(f"Source: {path}\n{text}")
        sources.append({"path": path, "score": score})

    context_text = "\n\n---\n\n".join(context_chunks)

    # Step 4: Build RAG prompt
    default_system = (
        "You are a helpful assistant that answers questions based on provided context. "
        "Use ONLY the information from the context to answer. "
        "If the context doesn't contain enough information to answer the question, say so. "
        "Cite sources when possible by mentioning the file path."
    )
    prompt = (
        f"Context:\n{context_text}\n\n"
        f"Question: {question}\n\n"
        "Answer the question based on the context above. Cite sources when helpful."
    )

    # Step 5: Call LLM for synthesis
    if llm_url:
        # External LLM (e.g., gateway)
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {llm_api_key}",
        }
        body = {
            "model": llm_model,
            "messages": [
                {"role": "system", "content": system_prompt or default_system},
                {"role": "user", "content": prompt},
            ],
            "temperature": 0.7,
        }

        client = session or httpx.AsyncClient(timeout=60.0)
        try:
            res = await client.post(llm_url, headers=headers, json=body)
            res.raise_for_status()
            llm_response = res.json()
            answer = (
                llm_response.get("choices", [{}])[0]
                .get("message", {})
                .get("content", "")
            )
        finally:
            if not session:
                await client.aclose()
    else:
        # Domo native LLM
        chat_body = {
            "input": prompt,
            "promptTemplate": {"template": "${input}"},
            "model": "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
        }

        llm_res = await ai_routes.llm_generate_text(
            text_input=prompt,
            auth=self.auth,
            chat_body=chat_body,
            return_raw=return_raw,
            context=context,
        )

        if return_raw:
            return llm_res

        answer = llm_res.response.get("output", "")

    return {
        "answer": answer,
        "sources": sources,
        "context": context_chunks if return_context else None,
    }

DomoFilesets dataclass

DomoFilesets(auth: DomoAuth)

Bases: DomoManager

Collection manager for DomoFileset objects.

create async

create(
    name: str,
    description: str | None = None,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoFileset | ResponseGetData

Create a new fileset and return a DomoFileset instance.

Parameters:

Name	Type	Description	Default
name	str	Name of the new fileset.	required
description	str | None	Optional description.	None
body	dict[str, Any] | None	Optional full request body; overrides name and description.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
DomoFileset | ResponseGetData	The created DomoFileset, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def create(
    self,
    name: str,
    description: str | None = None,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoFileset | ResponseGetData:
    """Create a new fileset and return a ``DomoFileset`` instance.

    Args:
        name: Name of the new fileset.
        description: Optional description.
        body: Optional full request body; overrides ``name`` and ``description``.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        The created ``DomoFileset``, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not body:
        body = {"name": name}
        if description is not None:
            body["description"] = description

    res = await fileset_routes.create_fileset(
        auth=self.auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    return DomoFileset.from_dict(auth=self.auth, obj=res.response)

get async

get(
    limit: int | None = None,
    offset: int | None = None,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoFileset] | ResponseGetData

Search or list filesets and return DomoFileset instances.

Parameters:

Name	Type	Description	Default
limit	int | None	Unused; accepted for interface compatibility.	None
offset	int | None	Unused; accepted for interface compatibility.	None
body	dict[str, Any] | None	Optional custom search body.	None
return_raw	bool	If True, return the raw ResponseGetData.	False
debug_api	bool	If True, print API debug info.	False
session	AsyncClient | None	Optional httpx.AsyncClient session.	None
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
list[DomoFileset] | ResponseGetData	A list of DomoFileset, or the raw ResponseGetData when return_raw.

Raises:

Type	Description
RouteError	If the underlying route request fails.

Source code in src/crew_dcs/classes/DomoFileset/__init__.py

async def get(
    self,
    limit: int | None = None,
    offset: int | None = None,
    body: dict[str, Any] | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoFileset] | ResponseGetData:
    """Search or list filesets and return ``DomoFileset`` instances.

    Args:
        limit: Unused; accepted for interface compatibility.
        offset: Unused; accepted for interface compatibility.
        body: Optional custom search body.
        return_raw: If True, return the raw ResponseGetData.
        debug_api: If True, print API debug info.
        session: Optional httpx.AsyncClient session.
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        A list of ``DomoFileset``, or the raw ``ResponseGetData`` when ``return_raw``.

    Raises:
        RouteError: If the underlying route request fails.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await fileset_routes.search_filesets(
        auth=self.auth,
        body=body,
        context=context,
    )

    if return_raw:
        return res

    filesets = res.response if isinstance(res.response, list) else []
    return [DomoFileset.from_dict(auth=self.auth, obj=fs) for fs in filesets]

Modules

• access_controller