user

user ¶

User Management Routes

This module provides comprehensive user management functionality for Domo instances, organized into three main categories:

user: Core user operations (get, search, create, update, delete)
attributes: User attribute management
properties: User property management and updates

All user-related functionality is accessible through this unified interface.

DeleteUserError ¶

DeleteUserError(
    res: ResponseGetData,
    message: str | None = None,
    user_id: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when user deletion operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    res: rgd.ResponseGetData,
    message: str | None = None,
    user_id: str | None = None,
    **kwargs,
):
    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

DownloadAvatar_Error ¶

DownloadAvatar_Error(
    user_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when user avatar download operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    user_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = "Avatar download failed"
    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

ResetPasswordPasswordUsedError ¶

ResetPasswordPasswordUsedError(
    user_id: str | None = None,
    res: ResponseGetData | None = None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when attempting to reset password to a previously used password.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    user_id: str | None = None,
    res: rgd.ResponseGetData | None = None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = "Password was previously used"
    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

SearchUserNotFoundError ¶

SearchUserNotFoundError(
    search_criteria: str,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when user search operations return no results.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self, search_criteria: str, res=None, message: str | None = None, **kwargs
):
    if not message:
        message = f"No users found matching: {search_criteria}"
    super().__init__(res=res, entity_id=search_criteria, message=message, **kwargs)

UserAttributes_CRUD_Error ¶

UserAttributes_CRUD_Error(
    operation: str,
    attribute_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when user attributes create, update, or delete operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    operation: str,
    attribute_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    if not message:
        message = f"User attributes {operation} operation failed"

    super().__init__(res=res, entity_id=attribute_id, message=message, **kwargs)

UserAttributes_GET_Error ¶

UserAttributes_GET_Error(
    attribute_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when user attributes retrieval operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    attribute_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    if not message:
        message = "User attributes retrieval failed"
    super().__init__(res=res, entity_id=attribute_id, message=message, **kwargs)

UserAttributes_IssuerType ¶

Bases: DomoEnumMixin, Enum

Types of user attribute issuers.

UserProperty ¶

UserProperty(
    property_type: UserProperty_Type,
    values: str | list[str],
)

Represents a user property with its type and values.

Initialize a user property.

Parameters:

Name	Type	Description	Default
`property_type`	`UserProperty_Type`	The type of property from UserProperty_Type enum	required
`values`	`str \| list[str]`	The value(s) for the property (can be single value or list)	required

Source code in src/crew_dcs/routes/user/properties.py

def __init__(self, property_type: UserProperty_Type, values: str | list[str]):
    """Initialize a user property.

    Args:
        property_type: The type of property from UserProperty_Type enum
        values: The value(s) for the property (can be single value or list)
    """
    self.property_type = property_type
    self.values = self._value_to_list(values)

to_dict ¶

to_dict()

Convert the property to dictionary format for API requests.

Returns:

Name	Type	Description
`dict`		Property in API format with key and values

Source code in src/crew_dcs/routes/user/properties.py

def to_dict(self):
    """Convert the property to dictionary format for API requests.

    Returns:
        dict: Property in API format with key and values
    """
    return {
        "key": self.property_type.value,
        "values": self._value_to_list(self.values),
    }

UserProperty_Type ¶

Bases: DomoEnumMixin, Enum

Enumeration of available user property types.

UserSharing_Error ¶

UserSharing_Error(
    operation: str,
    user_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when user sharing operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    operation: str,
    user_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs,
):
    if not message:
        message = f"User sharing {operation} failed"
    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

User_CRUD_Error ¶

User_CRUD_Error(
    operation: str,
    user_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

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

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self,
    operation: str,
    user_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs,
):
    # Use provided message if available, otherwise create default
    if not message:
        message = f"User {operation} operation failed"

    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

User_GET_Error ¶

User_GET_Error(
    user_id: str | None = None,
    res=None,
    message: str | None = None,
    **kwargs
)

Bases: RouteError

Raised when user retrieval operations fail.

Source code in src/crew_dcs/routes/user/exceptions.py

def __init__(
    self, user_id: str | None = None, res=None, message: str | None = None, **kwargs
):
    if not message:
        message = "User retrieval failed"

    super().__init__(res=res, entity_id=user_id, message=message, **kwargs)

clean_attribute_id ¶

clean_attribute_id(text: str) -> str

Clean attribute ID by removing non-alphanumeric characters.

Parameters:

Name	Type	Description	Default
`text`	`str`	Text to clean	required

Returns:

Name	Type	Description
`str`	`str`	Cleaned text with only alphanumeric characters

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

def clean_attribute_id(text: str) -> str:
    """Clean attribute ID by removing non-alphanumeric characters.

    Args:
        text: Text to clean

    Returns:
        str: Cleaned text with only alphanumeric characters
    """
    return re.sub(r"[^A-Za-z0-9]", "", text)

create_user `async` ¶

create_user(
    auth: DomoAuth,
    display_name: str,
    email_address: str,
    role_id: int,
    send_notification: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new user in the Domo instance.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`display_name`	`str`	Display name for the new user	required
`email_address`	`str`	Email address for the new user (must be valid)	required
`role_id`	`int`	Role ID to assign to the new user	required
`send_notification`	`bool`	Whether to send the invite email (v3 API `sendNotification` query param). Set False to suppress the invite email. The v1 API's `sendInvite` param is silently ignored by v3. Confirmed by live API testing in DUG Slack (2026-09-08, thread https://domousergroup.slack.com/archives/C012H05RWTC/p1788887597447669); runnable proof pending in knowledge-base/domo/gotchas/test_user_api_params.py.	`True`
`return_raw`	`bool`	Return raw API response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing created user information

Raises:

Type	Description
`User_CRUD_Error`	If user creation fails
`ValueError`	If email address is invalid

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_user(
    auth: DomoAuth,
    display_name: str,
    email_address: str,
    role_id: int,
    send_notification: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new user in the Domo instance.

    Args:
        auth: Authentication object
        display_name: Display name for the new user
        email_address: Email address for the new user (must be valid)
        role_id: Role ID to assign to the new user
        send_notification: Whether to send the invite email (v3 API
            `sendNotification` query param). Set False to suppress the
            invite email. The v1 API's `sendInvite` param is silently
            ignored by v3. Confirmed by live API testing in DUG Slack
            (2026-09-08, thread
            https://domousergroup.slack.com/archives/C012H05RWTC/p1788887597447669);
            runnable proof pending in
            knowledge-base/domo/gotchas/test_user_api_params.py.
        return_raw: Return raw API response without processing

    Returns:
        ResponseGetData object containing created user information

    Raises:
        User_CRUD_Error: If user creation fails
        ValueError: If email address is invalid
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v3/users"

    is_valid_email(email_address)

    body = {
        "displayName": display_name,
        "detail": {"email": email_address},
        "roleId": role_id,
    }

    res = await gd.get_data(
        url=url,
        method="POST",
        body=body,
        auth=auth,
        params={"sendNotification": send_notification},
        context=context,
    )

    if return_raw:
        return res

    if res.status == 400 and res.response == "Bad Request":
        raise User_CRUD_Error(
            operation="create",
            res=res,
            message=f"{res.response} - does this user {email_address} already exist?",
        )

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

    res.is_success = True
    return res

create_user_attribute `async` ¶

create_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    data_type: str | None = None,
    security_voter: str | None = None,
    issuer_type: UserAttributes_IssuerType | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new user attribute.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`attribute_id`	`str`	Unique identifier for the attribute (will be cleaned)	required
`name`	`str \| None`	Display name for the attribute (defaults to attribute_id)	`None`
`description`	`str \| None`	Description of the attribute (defaults to timestamp)	`None`
`data_type`	`str \| None`	Data type validator (default: "ANY_VALUE")	`None`
`security_voter`	`str \| None`	Security voter setting (default: "FULL_VIS_ADMIN_IDP")	`None`
`issuer_type`	`UserAttributes_IssuerType \| None`	Type of issuer (default: CUSTOM)	`None`
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming attribute creation

Raises:

Type	Description
`UserAttributes_CRUD_Error`	If attribute creation fails

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    data_type: str | None = None,
    security_voter: str | None = None,
    issuer_type: UserAttributes_IssuerType | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new user attribute.

    Args:
        auth: Authentication object
        attribute_id: Unique identifier for the attribute (will be cleaned)
        name: Display name for the attribute (defaults to attribute_id)
        description: Description of the attribute (defaults to timestamp)
        data_type: Data type validator (default: "ANY_VALUE")
        security_voter: Security voter setting (default: "FULL_VIS_ADMIN_IDP")
        issuer_type: Type of issuer (default: CUSTOM)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming attribute creation

    Raises:
        UserAttributes_CRUD_Error: If attribute creation fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    name = name or attribute_id
    attribute_id = clean_attribute_id(attribute_id)
    description = (
        description
        or f"updated via domolibrary {dt.datetime.now().strftime('%Y-%m-%d - %H:%M')}"
    )
    data_type = data_type or "ANY_VALUE"
    security_voter = security_voter or "FULL_VIS_ADMIN_IDP"
    issuer_type = issuer_type or UserAttributes_IssuerType.CUSTOM

    body = generate_create_user_attribute_body(
        attribute_id=attribute_id,
        issuer_type=issuer_type,
        name=name,
        data_type=data_type,
        security_voter=security_voter,
        description=description,
    )

    url = f"https://{auth.domo_instance}.domo.com/api/user/v1/properties/meta/keys/{attribute_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise UserAttributes_CRUD_Error(
            operation="create",
            attribute_id=attribute_id,
            message=f"Bad Request - does attribute {attribute_id} already exist?",
            res=res,
        )

    res.response = f"created {attribute_id}"

    return res

delete_user `async` ¶

delete_user(
    auth: DomoAuth,
    user_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a user from the Domo instance.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`user_id`	`str`	ID of the user to delete	required
`return_raw`	`bool`	Return raw API response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming user deletion

Raises:

Type	Description
`DeleteUserError`	If user deletion fails

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_user(
    auth: DomoAuth,
    user_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a user from the Domo instance.

    Args:
        auth: Authentication object
        user_id: ID of the user to delete
        return_raw: Return raw API response without processing

    Returns:
        ResponseGetData object confirming user deletion

    Raises:
        DeleteUserError: If user deletion fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/identity/v1/users/{user_id}"

    if context and context.debug_api:
        print(url)

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

    if return_raw:
        return res

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

    return res

delete_user_attribute `async` ¶

delete_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a user attribute.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`attribute_id`	`str`	ID of the attribute to delete	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming attribute deletion

Raises:

Type	Description
`UserAttributes_CRUD_Error`	If attribute deletion fails

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a user attribute.

    Args:
        auth: Authentication object
        attribute_id: ID of the attribute to delete
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming attribute deletion

    Raises:
        UserAttributes_CRUD_Error: If attribute deletion fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/user/v1/properties/meta/keys/{attribute_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise UserAttributes_CRUD_Error(
            operation="delete",
            attribute_id=attribute_id,
            message=f"Bad Request - failed to delete {attribute_id}",
            res=res,
        )

    res.response = f"deleted {attribute_id}"

    return res

download_avatar `async` ¶

download_avatar(
    auth: DomoAuth,
    user_id: str,
    pixels: int = 300,
    folder_path="./images",
    img_name=None,
    is_download_image: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Download a user's avatar image.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	ID of the user whose avatar to download	required
`auth`	`DomoAuth`	Authentication object	required
`pixels`	`int`	Size of the avatar in pixels (default: 300)	`300`
`folder_path`		Path to save the avatar image (default: "./images")	`'./images'`
`img_name`		Custom name for the image file (optional)	`None`
`is_download_image`	`bool`	Whether to save image to disk (default: True)	`True`
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
	ResponseGetData object containing avatar data

Raises:

Type	Description
`DownloadAvatar_Error`	If avatar download fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def download_avatar(
    auth: DomoAuth,
    user_id: str,
    pixels: int = 300,
    folder_path="./images",
    img_name=None,
    is_download_image: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Download a user's avatar image.

    Args:
        user_id: ID of the user whose avatar to download
        auth: Authentication object
        pixels: Size of the avatar in pixels (default: 300)
        folder_path: Path to save the avatar image (default: "./images")
        img_name: Custom name for the image file (optional)
        is_download_image: Whether to save image to disk (default: True)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing avatar data

    Raises:
        DownloadAvatar_Error: If avatar download fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/avatar/USER/{user_id}?size={pixels}"

    res = await gd.get_data_stream(
        url=url,
        method="GET",
        auth=auth,
        headers={"accept": "image/png;charset=utf-8"},
        context=context,
    )

    if return_raw:
        return res

    if res.status != 200:
        raise DownloadAvatar_Error(user_id=user_id, res=res)

    if is_download_image is True:
        img_name = f"{user_id}.png" if img_name is None else img_name

        if not os.path.exists(folder_path):
            os.makedirs(folder_path)

        with open(f"{folder_path}/{img_name}", "wb") as file:
            file.write(res.response)

    return res

generate_avatar_bytestr ¶

generate_avatar_bytestr(
    img_bytestr: bytes | str, img_type: str
) -> str

Generate base64 encoded avatar byte string for upload.

Parameters:

Name	Type	Description	Default
`img_bytestr`	`bytes \| str`	Image data as bytes or base64 string	required
`img_type`	`str`	Image type ('jpg' or 'png')	required

Returns:

Name	Type	Description
`str`	`str`	Base64 encoded image string with data URI prefix

Source code in src/crew_dcs/routes/user/properties.py

def generate_avatar_bytestr(img_bytestr: bytes | str, img_type: str) -> str:
    """Generate base64 encoded avatar byte string for upload.

    Args:
        img_bytestr: Image data as bytes or base64 string
        img_type: Image type ('jpg' or 'png')

    Returns:
        str: Base64 encoded image string with data URI prefix
    """
    if isinstance(img_bytestr, str):
        img_bytestr = img_bytestr.encode("utf-8")

    if not images.isBase64(img_bytestr):
        img_bytestr = base64.b64encode(img_bytestr)

    img_bytestr = img_bytestr.decode("utf-8")

    html_encoding = f"data:image/{img_type};base64,"

    if not img_bytestr.startswith(html_encoding):
        img_bytestr = html_encoding + img_bytestr

    return img_bytestr

generate_create_user_attribute_body ¶

generate_create_user_attribute_body(
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    issuer_type: UserAttributes_IssuerType | None = None,
    security_voter: str | None = None,
    data_type: str | None = None,
) -> dict

Generate request body for creating user attributes.

Parameters:

Name	Type	Description	Default
`attribute_id`	`str`	Unique identifier for the attribute	required
`name`	`str \| None`	Display name for the attribute	`None`
`description`	`str \| None`	Description of the attribute	`None`
`issuer_type`	`UserAttributes_IssuerType \| None`	Type of issuer for the attribute (UserAttributes_IssuerType enum)	`None`
`security_voter`	`str \| None`	Security voter setting	`None`
`data_type`	`str \| None`	Data type validator	`None`

Returns:

Name	Type	Description
`dict`	`dict`	Request body for attribute creation

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

def generate_create_user_attribute_body(
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    issuer_type: UserAttributes_IssuerType | None = None,
    security_voter: str | None = None,
    data_type: str | None = None,
) -> dict:
    """Generate request body for creating user attributes.

    Args:
        attribute_id: Unique identifier for the attribute
        name: Display name for the attribute
        description: Description of the attribute
        issuer_type: Type of issuer for the attribute (UserAttributes_IssuerType enum)
        security_voter: Security voter setting
        data_type: Data type validator

    Returns:
        dict: Request body for attribute creation
    """
    s = {"key": attribute_id}

    if issuer_type:
        s.update({"keyspace": issuer_type.value})

    if security_voter:
        s.update({"securityVoter": security_voter})

    if data_type:
        s.update({"validator": data_type})

    if name:
        s.update({"title": name})

    if description:
        s.update({"description": description})

    return s

generate_patch_user_property_body ¶

generate_patch_user_property_body(
    user_property_ls: list[UserProperty],
) -> dict

Generate request body for user property updates.

Parameters:

Name	Type	Description	Default
`user_property_ls`	`list[UserProperty]`	list of UserProperty objects to update	required

Returns:

Name	Type	Description
`dict`	`dict`	Request body with attributes array for PATCH request

Source code in src/crew_dcs/routes/user/properties.py

def generate_patch_user_property_body(user_property_ls: list[UserProperty]) -> dict:
    """Generate request body for user property updates.

    Args:
        user_property_ls: list of UserProperty objects to update

    Returns:
        dict: Request body with attributes array for PATCH request
    """
    return {
        "attributes": [user_property.to_dict() for user_property in user_property_ls]
    }

get_all_users `async` ¶

get_all_users(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Retrieve all users from Domo instance.

This function returns all users as provided by the Domo API endpoint /api/content/v2/users. By default, this typically includes active users and may exclude deleted users and Domo support accounts. The exact filtering behavior (e.g., inclusion of deleted or support users) depends on the API's implementation and may change over time. If you require information about deleted or support users, consult the Domo API documentation or inspect the raw response using return_raw=True. Args: auth: Authentication object return_raw: Return raw API response without processing

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing all users

Raises:

Type	Description
`User_GET_Error`	If user retrieval fails

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_all_users(
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve all users from Domo instance.

    This function returns all users as provided by the Domo API endpoint
    `/api/content/v2/users`. By default, this typically includes active users
    and may exclude deleted users and Domo support accounts. The exact
    filtering behavior (e.g., inclusion of deleted or support users) depends
    on the API's implementation and may change over time. If you require
    information about deleted or support users, consult the Domo API
    documentation or inspect the raw response using `return_raw=True`.
    Args:
        auth: Authentication object
        return_raw: Return raw API response without processing

    Returns:
        ResponseGetData object containing all users

    Raises:
        User_GET_Error: If user retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v2/users"

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

    if return_raw:
        return res

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

    return res

get_by_id `async` ¶

get_by_id(
    user_id: str,
    auth: DomoAuth,
    is_v2: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Retrieve a specific user by ID.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The unique identifier for the user	required
`auth`	`DomoAuth`	Authentication object	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`is_v2`	`bool`	If True, use v2 search API, otherwise use combined v2/v3 approach	`True`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing user information

Raises:

Type	Description
`User_GET_Error`	If user retrieval fails
`SearchUserNotFoundError`	If user with specified ID doesn't exist

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_by_id(
    user_id: str,
    auth: DomoAuth,
    is_v2: bool = True,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a specific user by ID.

    Args:
        user_id: The unique identifier for the user
        auth: Authentication object
        return_raw: Return raw API response without processing
        is_v2: If True, use v2 search API, otherwise use combined v2/v3 approach

    Returns:
        ResponseGetData object containing user information

    Raises:
        User_GET_Error: If user retrieval fails
        SearchUserNotFoundError: If user with specified ID doesn't exist
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not is_v2:
        return await _get_by_id(
            user_id=user_id,
            auth=auth,
            return_raw=return_raw,
            context=context,
        )

    res = await search_users_by_id(
        user_ids=[user_id],
        auth=auth,
        return_raw=return_raw,
        suppress_no_results_error=False,
        context=context,
    )

    if return_raw:
        return res

    res.response = res.response[0]
    return res

get_user_attribute_by_id `async` ¶

get_user_attribute_by_id(
    auth: DomoAuth,
    attribute_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve a specific user attribute by ID.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`attribute_id`	`str`	ID of the attribute to retrieve	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing the user attribute

Raises:

Type	Description
`UserAttributes_GET_Error`	If attribute retrieval fails or attribute not found

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_user_attribute_by_id(
    auth: DomoAuth,
    attribute_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve a specific user attribute by ID.

    Args:
        auth: Authentication object
        attribute_id: ID of the attribute to retrieve
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing the user attribute

    Raises:
        UserAttributes_GET_Error: If attribute retrieval fails or attribute not found
    """

    # Create a new context with incremented debug_num_stacks_to_drop for nested call
    context = RouteContext.build_context(context, **context_kwargs)

    res = await get_user_attributes(
        auth=auth,
        context=context,
        return_raw=return_raw,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise UserAttributes_GET_Error(attribute_id=attribute_id, res=res)

    res.response = next(
        (obj for obj in res.response if obj["key"] == attribute_id), None
    )

    if not res.response:
        raise UserAttributes_GET_Error(
            attribute_id=attribute_id,
            message=f"attribute {attribute_id} not found",
            res=res,
        )

    return res

get_user_attributes `async` ¶

get_user_attributes(
    auth: DomoAuth,
    issuer_type_ls: (
        list[UserAttributes_IssuerType] | None
    ) = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve user attributes from Domo instance.

User attributes can be of different types: IDP, domo-defined, or customer-defined. Note: API does not filter on the issuer type despite accepting filter parameter, so filtering is done client-side.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`issuer_type_ls`	`list[UserAttributes_IssuerType] \| None`	list of issuer types to retrieve (default: all types)	`None`
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing user attributes

Raises:

Type	Description
`UserAttributes_GET_Error`	If user attributes retrieval fails

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_user_attributes(
    auth: DomoAuth,
    issuer_type_ls: (
        list[UserAttributes_IssuerType] | None
    ) = None,  # use `UserAttributes_IssuerType` enum
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Retrieve user attributes from Domo instance.

    User attributes can be of different types: IDP, domo-defined, or customer-defined.
    Note: API does not filter on the issuer type despite accepting filter parameter,
    so filtering is done client-side.

    Args:
        auth: Authentication object
        issuer_type_ls: list of issuer types to retrieve (default: all types)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing user attributes

    Raises:
        UserAttributes_GET_Error: If user attributes retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    issuer_type_ls = issuer_type_ls or list(UserAttributes_IssuerType)

    issuer_types = ",".join([dmue.normalize_enum(member) for member in issuer_type_ls])

    params = {"issuerTypes": issuer_types}

    url = f"https://{auth.domo_instance}.domo.com/api/user/v1/properties/meta/keys"

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

    if return_raw:
        return res

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

    res.response = [
        obj
        for obj in res.response
        if obj["keyspace"] in [dmue.normalize_enum(member) for member in issuer_type_ls]
    ]

    return res

process_v1_search_users ¶

process_v1_search_users(
    v1_user_ls: list[dict],
) -> list[dict]

sanitizes the response from v1_users_search API and removes unnecessary attributes

Source code in src/crew_dcs/routes/user/core.py

def process_v1_search_users(
    v1_user_ls: list[dict],  # list of users from v1_users_search API
) -> list[dict]:  # sanitized list of users.
    """sanitizes the response from v1_users_search API and removes unnecessary attributes"""

    clean_users = []

    for obj in v1_user_ls:
        # dd_user = dd.DictDot(obj_user)

        attributes = obj.pop("attributes", [])

        clean_users.append(
            {**obj, **{attr["key"]: attr["values"][0] for attr in attributes}}
        )

    return clean_users

request_password_reset `async` ¶

request_password_reset(
    domo_instance: str,
    email: str,
    locale: str = "en-us",
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
)

Request a password reset for a user via email.

Parameters:

Name	Type	Description	Default
`domo_instance`	`str`	Name of the Domo instance	required
`email`	`str`	Email address of the user requesting password reset	required
`locale`	`str`	Locale for the reset email (default: "en-us")	`'en-us'`
`context`	`RouteContext \| None`	Optional RouteContext for request configuration	`None`
`return_raw`	`bool`	Return raw API response without processing	`False`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
	ResponseGetData object confirming password reset request

Raises:

Type	Description
`User_CRUD_Error`	If password reset request fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def request_password_reset(
    domo_instance: str,
    email: str,
    locale: str = "en-us",
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
):
    """Request a password reset for a user via email.

    Args:
        domo_instance: Name of the Domo instance
        email: Email address of the user requesting password reset
        locale: Locale for the reset email (default: "en-us")
        context: Optional RouteContext for request configuration
        return_raw: Return raw API response without processing
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming password reset request

    Raises:
        User_CRUD_Error: If password reset request fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{domo_instance}.domo.com/api/domoweb/auth/sendReset"

    params = {"email": email, "locale": locale}

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

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(
            operation="request_password_reset",
            res=res,
            message=f"unable to request password reset {res.response}",
        )

    return res

reset_password `async` ¶

reset_password(
    auth: DomoAuth,
    user_id: str,
    new_password: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Reset a user's password.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`user_id`	`str`	ID of the user whose password to reset	required
`new_password`	`str`	New password for the user	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming password reset

Raises:

Type	Description
`User_CRUD_Error`	If password reset fails
`ResetPasswordPasswordUsedError`	If password was previously used

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def reset_password(
    auth: DomoAuth,
    user_id: str,
    new_password: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Reset a user's password.

    Args:
        auth: Authentication object
        user_id: ID of the user whose password to reset
        new_password: New password for the user
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming password reset

    Raises:
        User_CRUD_Error: If password reset fails
        ResetPasswordPasswordUsedError: If password was previously used
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/identity/v1/password"

    body = {"domoUserId": user_id, "password": new_password}

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

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(
            operation="reset_password",
            user_id=user_id,
            res=res,
            message="unable to change password",
        )

    if (
        res.status == 200
        and res.response.get("description", None)
        == "Password has been used previously."
    ):
        raise ResetPasswordPasswordUsedError(
            user_id=user_id,
            res=res,
            message=res.response["description"].replace(".", ""),
        )

    return res

search_users `async` ¶

search_users(
    auth: DomoAuth,
    body: dict,
    loop_until_end: bool = True,
    limit: int = 200,
    maximum: int = 100,
    suppress_no_results_error: bool = False,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs
) -> ResponseGetData

Search users with flexible criteria using the v1 users search API.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`body`	`dict`	Search criteria body for the API request	required
`loop_until_end`	`bool`	Retrieve all available rows across all pages	`True`
`limit`	`int`	Maximum rows to return per request (pagination)	`200`
`maximum`	`int`	Maximum total rows to return (like SQL LIMIT)	`100`
`suppress_no_results_error`	`bool`	Don't raise error if no results found	`False`
`return_raw`	`bool`	Return raw API response without processing	`False`
`debug_loop`	`bool`	Enable loop debugging	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing search results

Raises:

Type	Description
`User_GET_Error`	If search request fails
`SearchUserNotFoundError`	If no users found and suppress_no_results_error is False

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_users(
    auth: DomoAuth,
    body: dict,
    loop_until_end: bool = True,  # retrieve all available rows
    limit: int = 200,  # maximum rows to return per request.  refers to PAGINATION
    maximum: int = 100,  # equivalent to the LIMIT or TOP clause in SQL, the number of rows to return total
    suppress_no_results_error: bool = False,
    debug_loop: bool = False,
    *,
    context: RouteContext | None = None,
    return_raw: bool = False,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Search users with flexible criteria using the v1 users search API.

    Args:
        auth: Authentication object
        body: Search criteria body for the API request
        loop_until_end: Retrieve all available rows across all pages
        limit: Maximum rows to return per request (pagination)
        maximum: Maximum total rows to return (like SQL LIMIT)
        suppress_no_results_error: Don't raise error if no results found
        return_raw: Return raw API response without processing
        debug_loop: Enable loop debugging

    Returns:
        ResponseGetData object containing search results

    Raises:
        User_GET_Error: If search request fails
        SearchUserNotFoundError: If no users found and suppress_no_results_error is False
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/identity/v1/users/search"

    offset_params = {"offset": "offset", "limit": "limit"}

    def body_fn(skip, limit, body):
        return {**body, "limit": limit, "offset": skip}

    def arr_fn(res: rgd.ResponseGetData):
        return res.response.get("users") if isinstance(res.response, dict) else []

    res = await gd.looper(
        auth=auth,
        method="POST",
        url=url,
        maximum=maximum,
        limit=limit,
        offset_params=offset_params,
        offset_params_in_body=True,
        loop_until_end=loop_until_end,
        arr_fn=arr_fn,
        body_fn=body_fn,
        body=body,
        debug_loop=debug_loop,
        context=context,
    )

    if return_raw:
        return res

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

    if not suppress_no_results_error and len(res.response) == 0:
        raise SearchUserNotFoundError(search_criteria=str(body), res=res)

    res.response = process_v1_search_users(res.response)

    return res

search_users_by_email `async` ¶

search_users_by_email(
    user_email_ls: list[str],
    auth: DomoAuth,
    return_raw: bool = False,
    suppress_no_results_error: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search for users by their email addresses using the v1 users search API.

Parameters:

Name	Type	Description	Default
`user_email_ls`	`list[str]`	list of user email addresses to search for	required
`auth`	`DomoAuth`	Authentication object	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`suppress_no_results_error`	`bool`	Don't raise error if no results found	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing found users

Raises:

Type	Description
`User_GET_Error`	If search request fails
`SearchUserNotFoundError`	If no users found and suppress_no_results_error is False

Note

Search does not appear to be case sensitive

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_users_by_email(
    user_email_ls: list[
        str
    ],  # list of user emails to search.  Note:  search does not appear to be case sensitive
    auth: DomoAuth,
    return_raw: bool = False,
    suppress_no_results_error: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:  # ResponseGetData with user list
    """Search for users by their email addresses using the v1 users search API.

    Args:
        user_email_ls: list of user email addresses to search for
        auth: Authentication object
        return_raw: Return raw API response without processing
        suppress_no_results_error: Don't raise error if no results found

    Returns:
        ResponseGetData object containing found users

    Raises:
        User_GET_Error: If search request fails
        SearchUserNotFoundError: If no users found and suppress_no_results_error is False

    Note:
        Search does not appear to be case sensitive
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_cn = dmce.chunk_list(user_email_ls, 1000)

    res_ls = await dmce.gather_with_concurrency(
        n=10,
        *[  # noqa: B026
            search_users(
                auth=auth,
                body={
                    # "showCount": true,
                    # "count": false,
                    "includeDeleted": False,
                    "includeSupport": False,
                    "limit": 200,
                    "offset": 0,
                    "sort": {"field": "displayName", "order": "ASC"},
                    "filters": [
                        {
                            "filterType": "text",
                            "field": "emailAddress",
                            "text": " ".join(user_ls),
                        }
                    ],
                    "parts": ["DETAILED", "GROUPS", "ROLE"],
                    "attributes": [
                        "id",
                        "displayName",
                        "roleId",
                        "department",
                        "title",
                        "emailAddress",
                        "phoneNumber",
                        "lastActivity",
                    ],
                },
                return_raw=return_raw,
                suppress_no_results_error=suppress_no_results_error,
                context=context,
            )
            for user_ls in user_cn
        ],
    )

    if return_raw:
        return res_ls[-1]

    res = res_ls[-1]

    res.response = [row for ls in [_.response for _ in res_ls] for row in ls]
    return res

search_users_by_id `async` ¶

search_users_by_id(
    auth: DomoAuth,
    user_ids: list[str],
    return_raw: bool = False,
    suppress_no_results_error: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Search for users by their IDs using the v1 users search API.

Parameters:

Name	Type	Description	Default
`user_ids`	`list[str]`	list of user IDs to search for	required
`auth`	`DomoAuth`	Authentication object	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`suppress_no_results_error`	`bool`	Don't raise error if no results found	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing found users

Raises:

Type	Description
`User_GET_Error`	If search request fails
`SearchUserNotFoundError`	If no users found and suppress_no_results_error is False

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_users_by_id(
    auth: DomoAuth,
    user_ids: list[str],  # list of user ids to search
    return_raw: bool = False,
    suppress_no_results_error: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:  # ResponseGetData with user list
    """Search for users by their IDs using the v1 users search API.

    Args:
        user_ids: list of user IDs to search for
        auth: Authentication object
        return_raw: Return raw API response without processing
        suppress_no_results_error: Don't raise error if no results found

    Returns:
        ResponseGetData object containing found users

    Raises:
        User_GET_Error: If search request fails
        SearchUserNotFoundError: If no users found and suppress_no_results_error is False
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_cn = dmce.chunk_list(user_ids, 1000)

    res_ls = await dmce.gather_with_concurrency(
        n=6,
        *[  # noqa: B026
            search_users(
                auth=auth,
                body={
                    # "showCount": true,
                    # "count": false,
                    "includeDeleted": False,
                    "includeSupport": False,
                    "filters": [
                        {
                            "field": "id",
                            "filterType": "value",
                            "values": user_ls,
                            "operator": "EQ",
                        }
                    ],
                    "parts": ["DETAILED", "GROUPS", "ROLE"],
                    "attributes": [
                        "id",
                        "displayName",
                        "roleId",
                        "department",
                        "title",
                        "emailAddress",
                        "phoneNumber",
                        "lastActivity",
                    ],
                },
                return_raw=return_raw,
                suppress_no_results_error=suppress_no_results_error,
                context=context,
            )
            for user_ls in user_cn
        ],
    )

    if return_raw:
        return res_ls[-1]

    res = res_ls[-1]

    res.response = [row for ls in [_.response for _ in res_ls] for row in ls]

    return res

search_virtual_user_by_subscriber_instance `async` ¶

search_virtual_user_by_subscriber_instance(
    auth: DomoAuth,
    subscriber_instance_ls: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve virtual users for subscriber instances tied to one publisher.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object for the publisher instance	required
`subscriber_instance_ls`	`list[str]`	list of subscriber Domo instance names	required
`return_raw`	`bool`	Return raw API response without processing	`False`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object containing virtual users

Raises:

Type	Description
`User_GET_Error`	If virtual user retrieval fails

Source code in src/crew_dcs/routes/user/core.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def search_virtual_user_by_subscriber_instance(
    auth: DomoAuth,  # domo auth object
    subscriber_instance_ls: list[str],  # list of subscriber domo instances
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:  # list of virtual domo users
    """Retrieve virtual users for subscriber instances tied to one publisher.

    Args:
        auth: Authentication object for the publisher instance
        subscriber_instance_ls: list of subscriber Domo instance names
        return_raw: Return raw API response without processing

    Returns:
        ResponseGetData object containing virtual users

    Raises:
        User_GET_Error: If virtual user retrieval fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/publish/v2/proxy_user/domain/"

    body = {
        "domains": [
            f"{subscriber_instance}.domo.com"
            for subscriber_instance in subscriber_instance_ls
        ]
    }

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

    if return_raw:
        return res

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

    return res

set_user_landing_page `async` ¶

set_user_landing_page(
    auth: DomoAuth,
    user_id: str,
    page_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Set a user's landing page.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`user_id`	`str`	ID of the user to update	required
`page_id`	`str`	ID of the page to set as landing page	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
	ResponseGetData object confirming the update

Raises:

Type	Description
`User_CRUD_Error`	If landing page update fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def set_user_landing_page(
    auth: DomoAuth,
    user_id: str,
    page_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Set a user's landing page.

    Args:
        auth: Authentication object
        user_id: ID of the user to update
        page_id: ID of the page to set as landing page
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming the update

    Raises:
        User_CRUD_Error: If landing page update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/landings/target/DESKTOP/entity/PAGE/id/{page_id}/{user_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(
            operation="set_landing_page",
            user_id=user_id,
            res=res,
        )

    return res

update_user `async` ¶

update_user(
    auth: DomoAuth,
    user_id: str,
    user_property_ls: list[UserProperty],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update user properties via the identity API.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	ID of the user to update	required
`user_property_ls`	`list[UserProperty]`	list of UserProperty objects with updates	required
`auth`	`DomoAuth`	Authentication object	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
	ResponseGetData object confirming property updates

Raises:

Type	Description
`User_CRUD_Error`	If property update fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_user(
    auth: DomoAuth,
    user_id: str,
    user_property_ls: list[UserProperty],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update user properties via the identity API.

    Args:
        user_id: ID of the user to update
        user_property_ls: list of UserProperty objects with updates
        auth: Authentication object
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming property updates

    Raises:
        User_CRUD_Error: If property update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/identity/v1/users/{user_id}"

    body = {}

    if isinstance(user_property_ls, list):
        if isinstance(user_property_ls[0], UserProperty):
            body = generate_patch_user_property_body(user_property_ls)

        if isinstance(user_property_ls[0], dict):
            body = {"attributes": user_property_ls}

    if not body:
        raise ValueError(f"Invalid user_property_ls format {user_property_ls}")

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

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(operation="update_properties", user_id=user_id, res=res)

    return res

update_user_attribute `async` ¶

update_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    issuer_type: UserAttributes_IssuerType = CUSTOM,
    data_type: str | None = None,
    security_voter: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update an existing user attribute.

The body must include all attribute parameters. This function will use the get_user_attribute_by_id function to retrieve existing values and merge them with the provided updates.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`attribute_id`	`str`	ID of the attribute to update	required
`name`	`str \| None`	Display name for the attribute	`None`
`description`	`str \| None`	Description of the attribute	`None`
`issuer_type`	`UserAttributes_IssuerType`	Type of issuer (default: CUSTOM)	`CUSTOM`
`data_type`	`str \| None`	Data type validator	`None`
`security_voter`	`str \| None`	Security voter setting	`None`
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming attribute update

Raises:

Type	Description
`UserAttributes_CRUD_Error`	If attribute update fails
`UserAttributes_GET_Error`	If attribute to update is not found

Source code in src/crew_dcs/routes/instance_config/user_attributes.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_user_attribute(
    auth: DomoAuth,
    attribute_id: str,
    name: str | None = None,
    description: str | None = None,
    issuer_type: UserAttributes_IssuerType = UserAttributes_IssuerType.CUSTOM,
    data_type: str | None = None,
    security_voter: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update an existing user attribute.

    The body must include all attribute parameters. This function will use the
    `get_user_attribute_by_id` function to retrieve existing values and merge
    them with the provided updates.

    Args:
        auth: Authentication object
        attribute_id: ID of the attribute to update
        name: Display name for the attribute
        description: Description of the attribute
        issuer_type: Type of issuer (default: CUSTOM)
        data_type: Data type validator
        security_voter: Security voter setting
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming attribute update

    Raises:
        UserAttributes_CRUD_Error: If attribute update fails
        UserAttributes_GET_Error: If attribute to update is not found
    """

    body = generate_create_user_attribute_body(
        attribute_id=attribute_id,
        issuer_type=issuer_type,
        name=name,
        description=description,
        data_type=data_type,
        security_voter=security_voter,
    )

    # Create a new context with incremented debug_num_stacks_to_drop for nested call
    context = RouteContext.build_context(context, **context_kwargs)

    res = await get_user_attribute_by_id(
        attribute_id=attribute_id,
        auth=auth,
        context=context,
        return_raw=return_raw,
    )

    if return_raw:
        return res

    body = {**res.response, **body}

    url = f"https://{auth.domo_instance}.domo.com/api/user/v1/properties/meta/keys/{attribute_id}"

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

    if return_raw:
        return res

    if not res.is_success:
        raise UserAttributes_CRUD_Error(
            operation="update",
            attribute_id=attribute_id,
            message=f"Bad Request - error updating {attribute_id}",
            res=res,
        )

    res.response = f"updated {attribute_id}"

    return res

upload_avatar `async` ¶

upload_avatar(
    auth: DomoAuth,
    user_id: int,
    img_bytestr: bytes,
    img_type: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Upload an avatar image for a user.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`user_id`	`int`	ID of the user to update avatar for	required
`img_bytestr`	`bytes`	Image data as bytes	required
`img_type`	`str`	Image type ('jpg' or 'png')	required
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
	ResponseGetData object confirming avatar upload

Raises:

Type	Description
`User_CRUD_Error`	If avatar upload fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upload_avatar(
    auth: DomoAuth,
    user_id: int,
    img_bytestr: bytes,
    img_type: str,  #'jpg or png'
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Upload an avatar image for a user.

    Args:
        auth: Authentication object
        user_id: ID of the user to update avatar for
        img_bytestr: Image data as bytes
        img_type: Image type ('jpg' or 'png')
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming avatar upload

    Raises:
        User_CRUD_Error: If avatar upload fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/avatar/bulk"

    body = {
        "base64Image": generate_avatar_bytestr(img_bytestr, img_type),
        "encodedImage": generate_avatar_bytestr(img_bytestr, img_type),
        "isOpen": False,
        "entityIds": [user_id],
        "entityType": "USER",
    }

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

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(
            operation="upload_avatar",
            user_id=str(user_id),
            res=res,
        )

    return res

user_is_allowed_direct_signon `async` ¶

user_is_allowed_direct_signon(
    auth: DomoAuth,
    user_ids: list[str],
    is_allow_dso: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Manage direct sign-on permissions for users.

Parameters:

Name	Type	Description	Default
`auth`	`DomoAuth`	Authentication object	required
`user_ids`	`list[str]`	list of user IDs to modify	required
`is_allow_dso`	`bool`	Whether to allow direct sign-on (default: True)	`True`
`return_raw`	`bool`	Return raw API response without processing	`False`
`context`	`RouteContext \| None`	RouteContext for request configuration	`None`
`**context_kwargs`		Additional context parameters (session, debug_api, etc.)	`{}`

Returns:

Type	Description
`ResponseGetData`	ResponseGetData object confirming permission changes

Raises:

Type	Description
`User_CRUD_Error`	If direct sign-on permission update fails

Source code in src/crew_dcs/routes/user/properties.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def user_is_allowed_direct_signon(
    auth: DomoAuth,
    user_ids: list[str],
    is_allow_dso: bool = True,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Manage direct sign-on permissions for users.

    Args:
        auth: Authentication object
        user_ids: list of user IDs to modify
        is_allow_dso: Whether to allow direct sign-on (default: True)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object confirming permission changes

    Raises:
        User_CRUD_Error: If direct sign-on permission update fails
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/content/v3/users/directSignOn"
    params = {"value": is_allow_dso}

    res = await gd.get_data(
        url=url,
        method="PUT",
        auth=auth,
        params=params,
        body=user_ids if isinstance(user_ids, list) else [user_ids],
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise User_CRUD_Error(
            operation="set_direct_signon",
            res=res,
        )

    return res

user

user ¶

DeleteUserError ¶

DownloadAvatar_Error ¶

ResetPasswordPasswordUsedError ¶

SearchUserNotFoundError ¶

UserAttributes_CRUD_Error ¶

UserAttributes_GET_Error ¶

UserAttributes_IssuerType ¶

UserProperty ¶

to_dict ¶

UserProperty_Type ¶

UserSharing_Error ¶

User_CRUD_Error ¶

User_GET_Error ¶

clean_attribute_id ¶

create_user async ¶

create_user_attribute async ¶

delete_user async ¶

delete_user_attribute async ¶

download_avatar async ¶

generate_avatar_bytestr ¶

generate_create_user_attribute_body ¶

generate_patch_user_property_body ¶

get_all_users async ¶

get_by_id async ¶

get_user_attribute_by_id async ¶

get_user_attributes async ¶

process_v1_search_users ¶

request_password_reset async ¶

reset_password async ¶

search_users async ¶

search_users_by_email async ¶

search_users_by_id async ¶

search_virtual_user_by_subscriber_instance async ¶

set_user_landing_page async ¶

update_user async ¶

update_user_attribute async ¶

upload_avatar async ¶

user_is_allowed_direct_signon async ¶

Modules¶

create_user `async` ¶

create_user_attribute `async` ¶

delete_user `async` ¶

delete_user_attribute `async` ¶

download_avatar `async` ¶

get_all_users `async` ¶

get_by_id `async` ¶

get_user_attribute_by_id `async` ¶

get_user_attributes `async` ¶

request_password_reset `async` ¶

reset_password `async` ¶

search_users `async` ¶

search_users_by_email `async` ¶

search_users_by_id `async` ¶

search_virtual_user_by_subscriber_instance `async` ¶

set_user_landing_page `async` ¶

update_user `async` ¶

update_user_attribute `async` ¶

upload_avatar `async` ¶

user_is_allowed_direct_signon `async` ¶