Function invite_user_to_organization

api/oss/src/routers/organization_router.py:170–245

Assigns a role to a user in an organization. Args: org_id (str): The ID of the organization. payload (InviteRequest): The payload containing the organization id, user email, and role to assign. workspace_id (str): The ID of the workspace. Returns: bool:

(
    org_id: str,
    payload: List[InviteRequest],
    workspace_id: str,
    request: Request,
)

    operation_id="invite_user_to_workspace",
)
async def invite_user_to_organization(
    org_id: str,
    payload: List[InviteRequest],
    workspace_id: str,
    request: Request,
):
    """
    Assigns a role to a user in an organization.

    Args:
        org_id (str): The ID of the organization.
        payload (InviteRequest): The payload containing the organization id, user email, and role to assign.
        workspace_id (str): The ID of the workspace.

    Returns:
        bool: True if the role was successfully assigned, False otherwise.

    Raises:
        HTTPException: If the user does not have permission to perform this action.
        HTTPException: If there is an error assigning the role to the user.
    """

    if len(payload) != 1:
        return JSONResponse(
            status_code=400,
            content={"detail": "Only one user can be invited at a time."},
        )

    if is_ee():
        user_org_workspace_data = await get_user_org_and_workspace_id(
            request.state.user_id
        )
        project = await db_manager_ee.get_project_by_workspace(workspace_id)
        has_permission = await check_rbac_permission(
            user_org_workspace_data=user_org_workspace_data,
            project_id=str(project.id),
            role=WorkspaceRole.WORKSPACE_ADMIN,
        )
        if not has_permission:
            return JSONResponse(
                status_code=403,
                content={
                    "detail": "You do not have permission to perform this action. Please contact your Organization Owner"
                },
            )

        owner = await db_manager.get_organization_owner(org_id)
        owner_domain = owner.email.split("@")[-1].lower() if owner else ""
        user_domain = payload[0].email.split("@")[-1].lower()
        skip_meter = owner_domain != "agenta.ai" and user_domain == "agenta.ai"

        if not skip_meter:
            check, _, _ = await check_entitlements(
                organization_id=request.state.organization_id,
                key=Gauge.USERS,
                delta=1,
            )