DomoUser

DomoUser

DomoUser module for managing Domo users.

CreateUser_MissingRoleError

CreateUser_MissingRoleError(domo_instance, email_address)

Bases: ClassError

Exception raised when role_id is missing during user creation.

Source code in src/crew_dcs/classes/DomoUser.py

def __init__(self, domo_instance, email_address):
    super().__init__(
        domo_instance=domo_instance,
        message=f"Error creating user {email_address}: missing role_id parameter",
    )

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)

DomoUser dataclass

DomoUser(
    auth: DomoAuth,
    id: str,
    raw: dict,
    display_name: str | None = None,
    email_address: str | None = None,
    role_id: int | None = None,
    department: str | None = None,
    title: str | None = None,
    avatar_key: str | None = None,
    avatar: Image | None = None,
    password: str | None = None,
    phone_number: str | None = None,
    web_landing_page: str | None = None,
    web_mobile_landing_page: str | None = None,
    employee_id: str | None = None,
    employee_number: str | None = None,
    hire_date: str | None = None,
    reports_to: str | None = None,
    publisher_domain: str | None = None,
    subscriber_domain: str | None = None,
    virtual_user_id: str | None = None,
    created_dt: datetime | None = None,
    last_activity_dt: datetime | None = None,
    custom_attributes: dict = dict(),
    domo_api_clients: list[Any] | None = None,
    domo_access_tokens: list[Any] | None = None,
    Role: Any | None = None,
    ApiClients: Any | None = None,
)

Bases: DomoEntity

A class for interacting with a Domo User

display_url property

display_url: str

Generate the URL to display this user in the Domo admin interface.

create async classmethod

create(
    auth: DomoAuth,
    display_name,
    email_address,
    role: Any,
    password: str | None = None,
    send_password_reset_email: bool = False,
    send_notification: bool = True,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

class method that creates a new Domo user.

Parameters:

Name	Type	Description	Default
send_notification	bool	Whether to send the invite email. Set False to suppress it (v3 API sendNotification query param). See routes/user/core.py::create_user for the v1 sendInvite gotcha and proof citation.	True

Source code in src/crew_dcs/classes/DomoUser.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    display_name,
    email_address,
    role: Any,  # DomoRole - will use default role if none provided
    password: str | None = None,
    send_password_reset_email: bool = False,
    send_notification: bool = True,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """class method that creates a new Domo user.

    Args:
        send_notification: Whether to send the invite email. Set False to
            suppress it (v3 API `sendNotification` query param). See
            routes/user/core.py::create_user for the v1 `sendInvite`
            gotcha and proof citation.
    """

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

    if not role:
        from .DomoInstanceConfig.role import DomoRoles

        role = await DomoRoles(auth=auth).get_default_role(context=context)

    res = await user_routes.create_user(
        auth=auth,
        display_name=display_name,
        email_address=email_address,
        role_id=role.id,
        send_notification=send_notification,
        context=context,
    )

    domo_user = await DomoUser.get_by_id(
        auth=auth,
        user_id=res.response.get("id") or res.response.get("userId"),
        context=context,
    )

    if password and domo_user:
        await domo_user.reset_password(new_password=password, context=context)

    elif send_password_reset_email and domo_user:
        await domo_user.request_password_reset(
            domo_instance=auth.domo_instance,
            email=domo_user.email_address,
            context=context,
        )

    return domo_user

download_avatar async

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

downloads a user's avatar to a folder and returns the byte representation of the image

Source code in src/crew_dcs/classes/DomoUser.py

async def download_avatar(
    self,
    pixels: int = 300,
    folder_path="./images",
    img_name=None,  # will default to user_id
    auth: DomoAuth | None = None,
    is_download_image: bool = True,  # option to prevent downloading the image file
    debug_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """downloads a user's avatar to a folder
    and returns the byte representation of the image
    """
    auth = auth or self.auth

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

    res = await user_routes.download_avatar(
        auth=self.auth,
        user_id=self.id,
        pixels=pixels,
        folder_path=folder_path,
        img_name=img_name,
        is_download_image=is_download_image,
        context=context,
    )

    if return_raw:
        return res

    self.avatar = ImageUtils.from_bytestr(data=res.response)

    return self.avatar

from_virtual_dict classmethod

from_virtual_dict(auth, obj: dict[str, Any])

Create DomoUser instance from virtual user JSON response.

Virtual users are created when content is published across instances.

Source code in src/crew_dcs/classes/DomoUser.py

@classmethod
def from_virtual_dict(cls, auth, obj: dict[str, Any]):
    """Create DomoUser instance from virtual user JSON response.

    Virtual users are created when content is published across instances.
    """
    return cls(
        id=obj["id"],
        auth=auth,
        publisher_domain=obj.get("publisherDomain"),
        subscriber_domain=obj.get("subscriberDomain"),
        virtual_user_id=obj.get("virtualUserId"),
        raw=obj,
    )

get_api_clients async

get_api_clients(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

retrieves Client_IDs for this user (assuming the authenticated user has manage rights). Note : the values will be masked, raw text values can only be retrieved via the UI

Source code in src/crew_dcs/classes/DomoUser.py

async def get_api_clients(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """
    retrieves Client_IDs for this user (assuming the authenticated user has manage rights).
    Note : the values will be masked, raw text values can only be retrieved via the UI
    """

    from .DomoInstanceConfig.api_client import ApiClients

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

    api_clients = ApiClients(auth=self.auth)

    res = await api_clients.get(
        context=context,
    )

    if return_raw:
        return res

    domo_clients = (
        [
            domo_client
            for domo_client in res.response
            if domo_client.domo_user.id == self.id  # type: ignore
        ]
        if isinstance(res, ResponseGetData)
        else []
    )

    if not domo_clients:
        print(
            f"Domo User {self.id} - {self.display_name} does not have any Client_IDs"
        )
        return False

    self.domo_api_clients = domo_clients

    return self.domo_api_clients

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    user_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve a Domo user by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
entity_id		User ID to retrieve	required
return_raw	bool	If True, return raw API response instead of DomoUser instance	False
debug_api	bool	Enable debug output for API call	False
debug_num_stacks_to_drop		Number of stack frames to drop in logging	2
session	AsyncClient | None	Optional httpx session for connection pooling	None
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	DomoUser instance or ResponseGetData if return_raw=True

Raises:

Type	Description
User_GET_Error	If user retrieval fails

Source code in src/crew_dcs/classes/DomoUser.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    user_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve a Domo user by ID.

    Args:
        auth: Authentication object
        entity_id: User ID to retrieve
        return_raw: If True, return raw API response instead of DomoUser instance
        debug_api: Enable debug output for API call
        debug_num_stacks_to_drop: Number of stack frames to drop in logging
        session: Optional httpx session for connection pooling
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoUser instance or ResponseGetData if return_raw=True

    Raises:
        User_GET_Error: If user retrieval fails
    """

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

    res = await user_routes.get_by_id(
        auth=auth,
        user_id=user_id,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        return None

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

    try:
        await domo_user.get_role(
            context=context,
        )
    except DomoError as e:
        print(e)

    return domo_user

request_password_reset async classmethod

request_password_reset(
    domo_instance: str,
    email: str,
    locale: str = "en-us",
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None
)

request password reset email. Note: does not require authentication.

Source code in src/crew_dcs/classes/DomoUser.py

@classmethod
async def request_password_reset(
    cls,
    domo_instance: str,
    email: str,
    locale: str = "en-us",
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
):
    """request password reset email.  Note: does not require authentication."""
    return await user_routes.request_password_reset(
        domo_instance=domo_instance,
        email=email,
        locale=locale,
        context=context,
        debug_api=debug_api,
        session=session,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        parent_class=cls.__name__,
    )

reset_password async

reset_password(
    new_password: str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

reset your password, will respect password restrictions set up in the Domo UI

Source code in src/crew_dcs/classes/DomoUser.py

async def reset_password(
    self: "DomoUser",
    new_password: str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """reset your password, will respect password restrictions set up in the Domo UI"""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await user_routes.reset_password(
        auth=self.auth,
        user_id=self.id,
        new_password=new_password,
        context=context,
    )

    self.password = new_password

    return res

DomoUser_NoSearchError

DomoUser_NoSearchError(
    cls_instance,
    message: str = "No users found matching search criteria",
    domo_instance: str | None = None,
)

Bases: ClassError

Exception raised when user search operations fail.

Source code in src/crew_dcs/classes/DomoUser.py

def __init__(
    self,
    cls_instance,
    message: str = "No users found matching search criteria",
    domo_instance: str | None = None,
):
    super().__init__(
        cls_instance=cls_instance,
        domo_instance=domo_instance,
        message=message,
    )

DomoUsers dataclass

DomoUsers(
    auth: DomoAuth,
    users: list[DomoUser] = list(),
    virtual_users: list[DomoUser] = list(),
)

Bases: DomoManager

a class for searching for Users

get async

get(
    *args,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **kwargs
) -> list[DomoUser]

retrieves all users from Domo

Source code in src/crew_dcs/classes/DomoUser.py

async def get(
    self,
    *args,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop=2,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **kwargs,
) -> list[DomoUser]:
    """retrieves all users from Domo"""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **kwargs,
    )

    res = await user_routes.get_all_users(
        auth=self.auth,
        context=context,
    )

    if return_raw:
        return res

    self.users = self._users_to_domo_user(user_ls=res.response, auth=self.auth)
    return self.users

upsert async

upsert(
    email_address: str,
    display_name: str | None = None,
    role: Any | None = None,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None
) -> DomoUser

Upsert a Domo user - update if exists, create if not.

Parameters:

Name	Type	Description	Default
email_address	str	User email address (required)	required
display_name	str | None	User display name (optional)	None
role	Any | None	DomoRole object (optional, will use default if not provided on creation)	None
debug_api	bool	Enable debug output for API calls	False
debug_num_stacks_to_drop	int	Number of stack frames to drop in logging	2
session	AsyncClient | None	Optional httpx session for connection pooling	None

Returns:

Name	Type	Description
DomoUser	DomoUser	The created or updated user

Raises:

Type	Description
ValidationError	If email address is invalid

Source code in src/crew_dcs/classes/DomoUser.py

@log_call(action_name="class")
async def upsert(
    self,
    email_address: str,
    display_name: str | None = None,
    role: Any | None = None,  # DomoRole
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
) -> DomoUser:
    """Upsert a Domo user - update if exists, create if not.

    Args:
        email_address: User email address (required)
        display_name: User display name (optional)
        role: DomoRole object (optional, will use default if not provided on creation)
        debug_api: Enable debug output for API calls
        debug_num_stacks_to_drop: Number of stack frames to drop in logging
        session: Optional httpx session for connection pooling

    Returns:
        DomoUser: The created or updated user

    Raises:
        ValidationError: If email address is invalid
    """
    is_valid_email(email_address)

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
    )

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
    )

    try:
        # Type narrowing: when only_allow_one=True and successful, returns DomoUser
        domo_user: DomoUser = await self.search_by_email(
            email=email_address,
            only_allow_one=True,
            context=context,
        )

        await logger.info(f"domo_user found {domo_user.id}")

        # Type guard to ensure we have a DomoUser instance
        if not isinstance(domo_user, DomoUser):
            raise ValueError(f"Expected DomoUser, got {type(domo_user)}")

        property_ls = []
        if display_name:
            property_ls.append(
                user_routes.UserProperty(
                    user_routes.UserProperty_Type.display_name, display_name
                )
            )

        if role and hasattr(role, "id") and role.id:
            property_ls.append(
                user_routes.UserProperty(
                    user_routes.UserProperty_Type.role_id, role.id
                )
            )

        if property_ls:
            await logger.info("Updating user properties for existing user.")
            await domo_user.update_properties(
                property_ls=property_ls,
                context=context,
            )
        return domo_user

    except (SearchUserNotFoundError, DomoUser_NoSearchError):
        # User doesn't exist, create new one

        await logger.info("User not found, creating new user.")

        created_user = await DomoUser.create(
            display_name=display_name
            or f"{email_address} - via dl {dt.date.today()}",
            email_address=email_address,
            auth=self.auth,
            role=role,
            context=context,
        )

        # Type guard for created user
        if created_user is None:
            raise ValueError(f"Failed to create user for email: {email_address}")  # noqa: B904

        await self.get(context=context)

        return created_user

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)

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)