filesets

filesets

Domo FileSet API routes.

Undocumented behaviour, confirmed live on datacrew-space:

Indexer classifies by UPLOADED FILENAME extension only. Content-Type and the fileType/name/path form fields are accepted, then ignored::

.md .json .txt  -> TEXT      COMPLETED  (searchable)
.yml .yaml      -> DOCUMENT  SKIPPED
.mdx            -> OTHER     SKIPPED

Upload of a SKIPPED type returns 200 and is never searchable. Store content under a .md path if it must be indexed.

File identity is (directoryPath, name) — no upsert. Replace = search by path, delete by id, upload.

Fileset_CRUD_Error

Fileset_CRUD_Error(
    operation: str,
    fileset_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when fileset create, update, or delete operations fail.

Source code in src/crew_dcs/routes/filesets.py

def __init__(
    self,
    operation: str,
    fileset_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"Fileset {operation} operation failed",
        entity_id=fileset_id,
        res=res,
        **kwargs,
    )

Fileset_GET_Error

Fileset_GET_Error(
    fileset_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when fileset retrieval operations fail.

Source code in src/crew_dcs/routes/filesets.py

def __init__(
    self,
    fileset_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or "Fileset retrieval failed",
        entity_id=fileset_id,
        res=res,
        **kwargs,
    )

create_data_file async

create_data_file(
    auth: DomoAuth,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new data file.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_data_file(
    auth: DomoAuth,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new data file."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="create data file", res=res)

    return res

create_fileset async

create_fileset(
    auth: DomoAuth,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_fileset(
    auth: DomoAuth,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="create", res=res)

    return res

create_fileset_folder async

create_fileset_folder(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a folder within a fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_fileset_folder(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a folder within a fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/files"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(
            operation="create folder", fileset_id=fileset_id, res=res
        )

    return res

create_filesets_index async

create_filesets_index(
    auth: DomoAuth,
    index_id: str,
    embedding_model: str = "domo.domo_ai.domo-embed-text-multilingual-v1:cohere",
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Creates a new vectorDB index.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_filesets_index(
    auth: DomoAuth,
    index_id: str,
    embedding_model: str = "domo.domo_ai.domo-embed-text-multilingual-v1:cohere",
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Creates a new vectorDB index."""

    url = f"{auth.domo_instance}.domo.com/api/recall/v1/indexes"

    payload = {
        "indexId": index_id,
        "embeddingModel": embedding_model,
    }
    res = await gd.get_data(
        url,
        method="POST",
        body=payload,
        auth=auth,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="create index", res=res)

    return res

delete_fileset async

delete_fileset(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a fileset by ID.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_fileset(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a fileset by ID."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="DELETE",
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="delete", fileset_id=fileset_id, res=res)

    return res

delete_fileset_file async

delete_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a file from a fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a file from a fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/files/{file_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="DELETE",
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(
            operation="delete file", fileset_id=fileset_id, res=res
        )

    return res

delete_fileset_file_by_path async

delete_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a file from a fileset by path.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a file from a fileset by path."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = (
        f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/path"
    )

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="DELETE",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(
            operation="delete file by path", fileset_id=fileset_id, res=res
        )

    return res

download_fileset_file async

download_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Download a file's binary content from a Domo fileset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
fileset_id	str	The ID of the fileset.	required
file_id	str	The ID of the file to download.	required
context	RouteContext | None	Optional route context.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the raw bytes content in the response field.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def download_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Download a file's binary content from a Domo fileset.

    Args:
        auth: Authentication object for Domo APIs.
        fileset_id: The ID of the fileset.
        file_id: The ID of the file to download.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the raw bytes content in the ``response`` field.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/files/{file_id}/download"

    # Build auth headers (accept any content type for binary downloads)
    headers = {
        "Connection": "keep-alive",
        "accept": "*/*",
    }
    if auth:
        headers.update(**auth.auth_header)

    request_metadata = rgd.RequestMetadata(
        url=url, headers=headers, body=None, params=None
    )

    # Use httpx directly to preserve raw bytes
    session, is_close_session = gd._create_httpx_session(
        session=context.session,
        is_verify=context.is_verify,
    )

    try:
        response = await session.request(
            method="GET",
            url=url,
            headers=headers,
            follow_redirects=context.is_follow_redirects,
            timeout=gd.DEFAULT_TIMEOUT,
        )
    except httpx.HTTPError:
        raise Fileset_GET_Error(fileset_id=fileset_id)  # noqa: B904
    finally:
        if is_close_session:
            await session.aclose()

    ok = 200 <= response.status_code <= 399
    response_headers = dict(response.headers) if response.headers else None
    elapsed = None
    try:
        if hasattr(response, "elapsed") and response.elapsed:
            elapsed = response.elapsed.total_seconds()
    except RuntimeError:
        elapsed = None

    if ok:
        result = rgd.ResponseGetData(
            status=response.status_code,
            response=response.content,
            is_success=True,
            request_metadata=request_metadata,
            response_headers=response_headers,
            elapsed=elapsed,
        )
    else:
        result = rgd.ResponseGetData.from_httpx_response(
            res=response,
            request_metadata=request_metadata,
        )
        raise Fileset_GET_Error(fileset_id=fileset_id, res=result)

    return result

download_fileset_file_by_path async

download_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    path: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Download a file's binary content from a Domo fileset by its path.

Mirrors download_fileset_file (raw bytes via httpx) but addresses the file by path instead of file id. Response behaviour is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
fileset_id	str	The ID of the fileset.	required
path	str	Path of the file within the fileset, e.g. dir/name.md.	required
context	RouteContext | None	Optional route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the raw bytes content in the response field.

Raises:

Type	Description
Fileset_GET_Error	If the request fails or returns a non-2xx status.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def download_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    path: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Download a file's binary content from a Domo fileset by its path.

    Mirrors ``download_fileset_file`` (raw bytes via httpx) but addresses the
    file by ``path`` instead of file id. Response behaviour is unverified
    against a live Domo.

    Args:
        auth: Authentication object for Domo APIs.
        fileset_id: The ID of the fileset.
        path: Path of the file within the fileset, e.g. ``dir/name.md``.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the raw bytes content in the ``response`` field.

    Raises:
        Fileset_GET_Error: If the request fails or returns a non-2xx status.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/path/download"
    params = {"path": path}

    headers = {
        "Connection": "keep-alive",
        "accept": "*/*",
    }
    if auth:
        headers.update(**auth.auth_header)

    request_metadata = rgd.RequestMetadata(
        url=url, headers=headers, body=None, params=params
    )

    # Use httpx directly to preserve raw bytes
    session, is_close_session = gd._create_httpx_session(
        session=context.session,
        is_verify=context.is_verify,
    )

    try:
        response = await session.request(
            method="GET",
            url=url,
            headers=headers,
            params=params,
            follow_redirects=context.is_follow_redirects,
            timeout=gd.DEFAULT_TIMEOUT,
        )
    except httpx.HTTPError:
        raise Fileset_GET_Error(fileset_id=fileset_id)  # noqa: B904
    finally:
        if is_close_session:
            await session.aclose()

    ok = 200 <= response.status_code <= 399
    response_headers = dict(response.headers) if response.headers else None
    elapsed = None
    try:
        if hasattr(response, "elapsed") and response.elapsed:
            elapsed = response.elapsed.total_seconds()
    except RuntimeError:
        elapsed = None

    if not ok:
        result = rgd.ResponseGetData.from_httpx_response(
            res=response,
            request_metadata=request_metadata,
        )
        raise Fileset_GET_Error(fileset_id=fileset_id, res=result)

    return rgd.ResponseGetData(
        status=response.status_code,
        response=response.content,
        is_success=True,
        request_metadata=request_metadata,
        response_headers=response_headers,
        elapsed=elapsed,
    )

embed_image async

embed_image(
    auth: DomoAuth,
    body: dict | None = None,
    image_data: str | None = None,
    media_type: str | None = None,
    data_type: EmbedData_Type = "base64",
    model: str = "domo.domo_ai",
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create an embedding for a base64 encoded image using Domo's AI services.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def embed_image(
    auth: DomoAuth,
    body: dict | None = None,
    image_data: str | None = None,
    media_type: str | None = None,
    data_type: EmbedData_Type = "base64",
    model: str = "domo.domo_ai",
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Create an embedding for a base64 encoded image using Domo's AI services.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Utility function is_valid_base64_image should be called by the orchestrator before this.
    # This route function assumes valid base64 data.

    api_url = f"https://{auth.domo_instance}.domo.com/api/ai/v1/embedding/image"

    body = body or {
        "input": [{"type": "", "mediaType": "", "data": ""}],
        "model": "",
    }
    if image_data:
        body["input"][0].update({"data": image_data})

    if media_type:
        body["input"][0].update({"mediaType": media_type})

    if data_type:
        body["input"][0].update({"type": data_type})

    if model:
        body.update({"model": model})

    res = await gd.get_data(
        auth=auth,
        url=api_url,
        method="POST",
        body=body,
        context=context,
    )
    if not res.is_success:
        raise Fileset_CRUD_Error(operation="embed image", res=res)

    return res

get_data_file_by_id async

get_data_file_by_id(
    auth: DomoAuth,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieves the content of a data file from Domo.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_data_file_by_id(
    auth: DomoAuth,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieves the content of a data file from Domo.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files/{file_id}"
    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )
    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=file_id, res=res)
    return res

get_data_file_details async

get_data_file_details(
    auth: DomoAuth,
    file_id: str,
    expand: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Retrieve the details of a Domo data file.

Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
file_id	str	The ID of the data file.	required
expand	str | None	Optional expand query value (the register example is revisions); only sent when given.	None
context	RouteContext | None	Optional route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the file details in the response field.

Raises:

Type	Description
Fileset_GET_Error	If the request returns a non-2xx status.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_data_file_details(
    auth: DomoAuth,
    file_id: str,
    expand: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Retrieve the details of a Domo data file.

    Response shape is unverified against a live Domo.

    Args:
        auth: Authentication object for Domo APIs.
        file_id: The ID of the data file.
        expand: Optional ``expand`` query value (the register example is
            ``revisions``); only sent when given.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the file details in the ``response`` field.

    Raises:
        Fileset_GET_Error: If the request returns a non-2xx status.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files/{file_id}/details"
    params = {"expand": expand} if expand is not None else None

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        params=params,
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=file_id, res=res)

    return res

get_data_file_revision async

get_data_file_revision(
    auth: DomoAuth,
    file_id: str,
    revision_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Retrieve a specific revision of a Domo data file.

Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
file_id	str	The ID of the data file.	required
revision_id	str	The ID of the revision.	required
context	RouteContext | None	Optional route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the revision in the response field.

Raises:

Type	Description
Fileset_GET_Error	If the request returns a non-2xx status.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_data_file_revision(
    auth: DomoAuth,
    file_id: str,
    revision_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Retrieve a specific revision of a Domo data file.

    Response shape is unverified against a live Domo.

    Args:
        auth: Authentication object for Domo APIs.
        file_id: The ID of the data file.
        revision_id: The ID of the revision.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the revision in the ``response`` field.

    Raises:
        Fileset_GET_Error: If the request returns a non-2xx status.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files/{file_id}/revisions/{revision_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=file_id, res=res)

    return res

get_data_file_revision_details async

get_data_file_revision_details(
    auth: DomoAuth,
    file_id: str,
    revision_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Retrieve the details of a specific revision of a Domo data file.

Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
file_id	str	The ID of the data file.	required
revision_id	str	The ID of the revision.	required
context	RouteContext | None	Optional route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the revision details in the response field.

Raises:

Type	Description
Fileset_GET_Error	If the request returns a non-2xx status.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_data_file_revision_details(
    auth: DomoAuth,
    file_id: str,
    revision_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Retrieve the details of a specific revision of a Domo data file.

    Response shape is unverified against a live Domo.

    Args:
        auth: Authentication object for Domo APIs.
        file_id: The ID of the data file.
        revision_id: The ID of the revision.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the revision details in the ``response`` field.

    Raises:
        Fileset_GET_Error: If the request returns a non-2xx status.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files/{file_id}/revisions/{revision_id}/details"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=file_id, res=res)

    return res

get_fileset_access async

get_fileset_access(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get the access list for a fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_fileset_access(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get the access list for a fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/access"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=fileset_id, res=res)

    return res

get_fileset_file async

get_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a file from a fileset by file ID.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a file from a fileset by file ID."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/files/{file_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=fileset_id, res=res)

    return res

get_fileset_file_by_path async

get_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a file from a fileset by path.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_fileset_file_by_path(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a file from a fileset by path."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = (
        f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/path"
    )

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=fileset_id, res=res)

    return res

get_fileset_stats async

get_fileset_stats(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get file type counts and index status for a fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_fileset_stats(
    auth: DomoAuth,
    fileset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get file type counts and index status for a fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/stats"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=fileset_id, res=res)

    return res

get_recall_index_access async

get_recall_index_access(
    auth: DomoAuth,
    index_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

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

Lives here beside create_filesets_index, the existing recall/v1 route. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
index_id	str	The ID of the Recall index.	required
context	RouteContext | None	Optional route context.	None
**context_kwargs	Any	Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the access list in the response field.

Raises:

Type	Description
Fileset_GET_Error	If the request returns a non-2xx status.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_recall_index_access(
    auth: DomoAuth,
    index_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the access list for a Recall (vectorDB) index.

    Lives here beside ``create_filesets_index``, the existing ``recall/v1``
    route. Response shape is unverified against a live Domo.

    Args:
        auth: Authentication object for Domo APIs.
        index_id: The ID of the Recall index.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the access list in the ``response`` field.

    Raises:
        Fileset_GET_Error: If the request returns a non-2xx status.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = (
        f"https://{auth.domo_instance}.domo.com/api/recall/v1/indexes/{index_id}/access"
    )

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=index_id, res=res)

    return res

query_fileset async

query_fileset(
    auth: DomoAuth,
    fileset_id: str,
    query: str,
    directory_path: str | None = None,
    top_k: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Perform an AI search query across indexed files in a fileset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
fileset_id	str	The ID of the fileset to query.	required
query	str	The search text.	required
directory_path	str | None	Optional directory path to scope the search.	None
top_k	int | None	Optional maximum number of results to return.	None
context	RouteContext | None	Optional route context.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the query results.

Shape::

{"matches": [{"content":  {"text", "type"},
              "metadata": {"fileId", "path"},
              "score":    0.6334517}]}

score is a float ~0-1. Metric undocumented — thresholds tuned on another embedding model do NOT transfer; re-derive per corpus.

directory_path -> API pathPrefixFilter. Scopes retrieval to a sub-tree, but access is granted per FILESET (get_fileset_access) — a prefix narrows results, it does not isolate permissions.

SKIPPED files never appear in matches (see module docstring).

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def query_fileset(
    auth: DomoAuth,
    fileset_id: str,
    query: str,
    directory_path: str | None = None,
    top_k: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Perform an AI search query across indexed files in a fileset.

    Args:
        auth: Authentication object for Domo APIs.
        fileset_id: The ID of the fileset to query.
        query: The search text.
        directory_path: Optional directory path to scope the search.
        top_k: Optional maximum number of results to return.
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the query results.

    Shape::

        {"matches": [{"content":  {"text", "type"},
                      "metadata": {"fileId", "path"},
                      "score":    0.6334517}]}

    `score` is a float ~0-1. Metric undocumented — thresholds tuned on another
    embedding model do NOT transfer; re-derive per corpus.

    `directory_path` -> API `pathPrefixFilter`. Scopes retrieval to a sub-tree,
    but access is granted per FILESET (`get_fileset_access`) — a prefix narrows
    results, it does not isolate permissions.

    SKIPPED files never appear in `matches` (see module docstring).
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/query"

    body: dict[str, Any] = {"query": query}
    if directory_path is not None:
        body["directoryPath"] = directory_path
    if top_k is not None:
        body["topK"] = top_k

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(fileset_id=fileset_id, res=res)

    return res

search_fileset_files async

search_fileset_files(
    auth: DomoAuth,
    domo_fileset_id: str,
    body: dict | None = None,
    directory_path: str = "",
    immediate_children: bool = True,
    loop_until_end: bool = False,
    limit: int = 100,
    maximum: int = 0,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search files within a fileset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	DomoAuth object.	required
domo_fileset_id	str	The fileset ID to search.	required
body	dict | None	Optional custom search body.	None
directory_path	str	Directory path to scope the search.	''
immediate_children	bool	If True, return only immediate children; if False, return all files recursively.	True
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
debug_loop	bool	Enable loop debugging.	False
context	RouteContext | None	Optional RouteContext.	None

Returns:

Type	Description
ResponseGetData	ResponseGetData with files in response.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_fileset_files(
    auth: DomoAuth,
    domo_fileset_id: str,
    body: dict | None = None,
    directory_path: str = "",
    immediate_children: bool = True,
    loop_until_end: bool = False,
    limit: int = 100,
    maximum: int = 0,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search files within a fileset.

    Args:
        auth: DomoAuth object.
        domo_fileset_id: The fileset ID to search.
        body: Optional custom search body.
        directory_path: Directory path to scope the search.
        immediate_children: If True, return only immediate children; if False, return all files recursively.
        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).
        debug_loop: Enable loop debugging.
        context: Optional RouteContext.

    Returns:
        ResponseGetData with files in response.
    """
    base_url = (
        f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{domo_fileset_id}/files/search"
        f"?directoryPath={directory_path}&immediateChildren={str(immediate_children).lower()}&limit={limit}"
    )

    if not body:
        body = {
            "fieldSort": [{"field": "created", "order": "DESC"}],
            "filters": [],
            "dateFilters": [],
        }

    if not loop_until_end:
        res = await gd.get_data(
            auth=auth,
            method="POST",
            url=base_url,
            body=body,
            context=context,
        )
        if not res.is_success:
            raise Fileset_GET_Error(fileset_id=domo_fileset_id, res=res)
        return res

    # Cursor-based pagination via looper
    # The fileset search API's cursor is a QUERY PARAM (`next=`) on the URL,
    # not a field in the POST body — passing it as `body.pageContext.next`
    # is silently ignored by the API and every page repeats page 1. Confirmed
    # live against the domo-documentation-hub fileset: identical `files[0]`
    # and identical (never-advancing) `pageContext.next` on every page when
    # the cursor was sent in the body.
    all_files: list[dict] = []
    next_cursor: str | None = None
    prev_cursor: str | None = None
    res = None

    while True:
        url = (
            base_url
            if not next_cursor
            else f"{base_url}&next={quote(next_cursor, safe='')}"
        )
        res = await gd.get_data(
            auth=auth,
            method="POST",
            url=url,
            body=body,
            context=context,
        )
        if not res.is_success:
            raise Fileset_GET_Error(fileset_id=domo_fileset_id, res=res)

        page_files = (
            res.response.get("files", [])
            if isinstance(res.response, dict)
            else res.response
            if isinstance(res.response, list)
            else []
        )
        all_files.extend(page_files)

        if debug_loop:
            print(
                f"🐛 search_fileset_files: {len(all_files)} files so far (page: {len(page_files)})"
            )

        # Check for cursor
        page_ctx = (
            res.response.get("pageContext", {})
            if isinstance(res.response, dict)
            else {}
        )
        next_cursor = page_ctx.get("next")
        if not next_cursor or next_cursor == prev_cursor:
            break

        # Check maximum
        if maximum and len(all_files) >= maximum:
            all_files = all_files[:maximum]
            break

        prev_cursor = next_cursor

    return rgd.ResponseGetData(
        status=res.status,
        response=all_files,
        is_success=True,
        request_metadata=res.request_metadata if res else None,
        additional_information=res.additional_information if res else None,
    )

search_filesets async

search_filesets(
    auth: DomoAuth,
    body: dict[str, Any] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search/list filesets.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_filesets(
    auth: DomoAuth,
    body: dict[str, Any] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search/list filesets."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/search"

    if not body:
        body = {
            "fieldSort": [{"field": "created", "order": "DESC"}],
            "filters": [],
            "dateFilters": [],
        }

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_GET_Error(res=res)

    return res

share_fileset async

share_fileset(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Add access entries to a fileset. The body must contain a 'fileSetAccess' key with a list of access entries. Each entry has entityId (int), entityType ('USER'|'GROUP'), and permission ('READ'). Cannot include OWNER permission entries.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def share_fileset(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Add access entries to a fileset. The body must contain a 'fileSetAccess' key
    with a list of access entries. Each entry has entityId (int), entityType ('USER'|'GROUP'),
    and permission ('READ'). Cannot include OWNER permission entries."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/access"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="share", fileset_id=fileset_id, res=res)

    return res

transfer_fileset_ownership async

transfer_fileset_ownership(
    auth: DomoAuth,
    fileset_id: str,
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Transfer ownership of a fileset to a different user.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def transfer_fileset_ownership(
    auth: DomoAuth,
    fileset_id: str,
    user_id: int,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Transfer ownership of a fileset to a different user."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/ownership"

    body = {"userId": user_id}

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(
            operation="transfer ownership", fileset_id=fileset_id, res=res
        )

    return res

update_data_file async

update_data_file(
    auth: DomoAuth,
    file_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update an existing data file.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_data_file(
    auth: DomoAuth,
    file_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update an existing data file."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/data/v1/data-files/{file_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="PUT",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(
            operation="update data file", fileset_id=file_id, res=res
        )

    return res

update_fileset async

update_fileset(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update a fileset.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_fileset(
    auth: DomoAuth,
    fileset_id: str,
    body: dict[str, Any],
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update a fileset."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Fileset_CRUD_Error(operation="update", fileset_id=fileset_id, res=res)

    return res

upload_fileset_file async

upload_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_name: str,
    file_content: str | bytes,
    directory_path: str = "",
    content_type: str = "text/plain",
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Upload a file to a Domo fileset using multipart/form-data.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for Domo APIs.	required
fileset_id	str	The ID of the fileset to upload to.	required
file_name	str	Name for the uploaded file.	required
file_content	str | bytes	The file content (string or bytes).	required
directory_path	str	Directory path within the fileset (default: "").	''
content_type	str	MIME type for the file (default: "text/plain").	'text/plain'
context	RouteContext | None	Optional route context.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData with the upload result.

content_type does not affect indexing — only file_name's extension does (module docstring). .mdx + content_type="text/markdown" still lands SKIPPED, and still returns 200. Silent failure.

Uploading over an existing (directory_path, file_name) duplicates rather than replaces. Delete first.

Source code in src/crew_dcs/routes/filesets.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_fileset_file(
    auth: DomoAuth,
    fileset_id: str,
    file_name: str,
    file_content: str | bytes,
    directory_path: str = "",
    content_type: str = "text/plain",
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Upload a file to a Domo fileset using multipart/form-data.

    Args:
        auth: Authentication object for Domo APIs.
        fileset_id: The ID of the fileset to upload to.
        file_name: Name for the uploaded file.
        file_content: The file content (string or bytes).
        directory_path: Directory path within the fileset (default: "").
        content_type: MIME type for the file (default: "text/plain").
        context: Optional route context.
        **context_kwargs: Additional context parameters.

    Returns:
        ResponseGetData with the upload result.

    `content_type` does not affect indexing — only `file_name`'s extension does
    (module docstring). `.mdx` + `content_type="text/markdown"` still lands
    SKIPPED, and still returns 200. Silent failure.

    Uploading over an existing `(directory_path, file_name)` duplicates rather
    than replaces. Delete first.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    url = f"https://{auth.domo_instance}.domo.com/api/files/v1/filesets/{fileset_id}/files"

    # Build auth headers (without Content-Type; httpx sets multipart boundary automatically)
    headers = {
        "Connection": "keep-alive",
        "accept": "application/json, text/plain",
    }
    if auth:
        headers.update(**auth.auth_header)

    # Build the multipart form fields.
    #
    # NOTE: this used to send a single `createFileRequest` field containing
    # `{"directoryPath": ...}` as a JSON string. That shape uploads
    # successfully (200) but the API silently classifies the file as
    # `fileType: "OTHER"` / `indexStatus: "SKIPPED"` ("File type is not
    # supported") — confirmed live against the domo-documentation-hub
    # fileset. The working shape (confirmed against the same fileset via
    # the `Domo Docs Sync` CodeEngine package, which reliably lands files as
    # `indexStatus: "COMPLETED"`) is four flat form fields, not one JSON
    # blob: name, path, directoryPath, fileType — where fileType is the bare
    # extension (e.g. "md", "mdx"), not a MIME type.
    path = f"{directory_path.strip('/')}/{file_name}" if directory_path else file_name
    file_type = file_name.rsplit(".", 1)[-1].lower() if "." in file_name else ""

    files = {
        "file": (file_name, file_content, content_type),
    }
    form_data = {
        "name": file_name,
        "path": path,
        "directoryPath": directory_path,
        "fileType": file_type,
    }

    request_metadata = rgd.RequestMetadata(
        url=url, headers=headers, body=None, params=None
    )

    # Use httpx directly for multipart upload
    session, is_close_session = gd._create_httpx_session(
        session=context.session,
        is_verify=context.is_verify,
    )

    try:
        response = await session.request(
            method="POST",
            url=url,
            headers=headers,
            files=files,
            data=form_data,
            follow_redirects=context.is_follow_redirects,
            timeout=gd.DEFAULT_TIMEOUT,
        )
    except httpx.HTTPError:
        raise Fileset_CRUD_Error(operation="upload file", fileset_id=fileset_id)  # noqa: B904
    finally:
        if is_close_session:
            await session.aclose()

    result = rgd.ResponseGetData.from_httpx_response(
        res=response,
        request_metadata=request_metadata,
    )

    if not result.is_success:
        raise Fileset_CRUD_Error(
            operation="upload file", fileset_id=fileset_id, res=result
        )

    return result