Invite users to the workspace programmatically

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ### Features
 - Series objects accept `timestamps` and `steps` in their constructors ([#1318](https://github.com/neptune-ai/neptune-client/pull/1318))
+- Users can be invited to the workspace with `management` api ([#1333](https://github.com/neptune-ai/neptune-client/pull/1333))
 
 ## neptune 1.1.1
 

src/neptune/management/__init__.py

@@ -100,6 +100,7 @@
 See also the API reference in the docs: https://docs.neptune.ai/api/management
 """
 from .internal.api import (
+    WorkspaceMemberRole,
     add_project_member,
     add_project_service_account,
     create_project,
@@ -109,6 +110,7 @@
     get_project_service_account_list,
     get_workspace_member_list,
     get_workspace_service_account_list,
+    invite_to_workspace,
     remove_project_member,
     remove_project_service_account,
     trash_objects,
@@ -126,6 +128,8 @@
     "add_project_member",
     "remove_project_member",
     "get_workspace_member_list",
+    "invite_to_workspace",
+    "WorkspaceMemberRole",
     "add_project_service_account",
     "remove_project_service_account",
     "get_project_service_account_list",

src/neptune/management/exceptions.py

@@ -168,3 +168,13 @@ class IncorrectIdentifierException(ManagementOperationFailure):
 class ObjectNotFound(ManagementOperationFailure):
     code = 22
     description = "Object not found."
+
+
+class WorkspaceOrUserNotFound(ManagementOperationFailure):
+    code = 23
+    description = "Workspace '{workspace}' or user '{user}' could not be found."
+
+
+class UserAlreadyInvited(ManagementOperationFailure):
+    code = 24
+    description = "User '{user}' has already been invited to the workspace '{workspace}'."

src/neptune/management/internal/api.py

@@ -22,6 +22,8 @@
     "add_project_member",
     "remove_project_member",
     "get_workspace_member_list",
+    "invite_to_workspace",
+    "WorkspaceMemberRole",
     "add_project_service_account",
     "remove_project_service_account",
     "get_project_service_account_list",
@@ -79,8 +81,10 @@
     ServiceAccountNotExistsOrWithoutAccess,
     ServiceAccountNotFound,
     UserAlreadyHasAccess,
+    UserAlreadyInvited,
     UserNotExistsOrWithoutAccess,
     WorkspaceNotFound,
+    WorkspaceOrUserNotFound,
 )
 from neptune.management.internal.dto import (
     ProjectMemberRoleDTO,
@@ -90,6 +94,7 @@
 )
 from neptune.management.internal.types import ProjectVisibility
 from neptune.management.internal.utils import (
+    WorkspaceMemberRole,
     extract_project_and_workspace,
     normalize_project_name,
 )
@@ -563,6 +568,101 @@ def get_workspace_service_account_list(workspace: str, *, api_token: Optional[st
     }
 
 
+@with_api_exceptions_handler
+def invite_to_workspace(
+    *,
+    username: Optional[str] = None,
+    email: Optional[str] = None,
+    workspace: str,
+    api_token: Optional[str] = None,
+    role: Union[WorkspaceMemberRole, str] = WorkspaceMemberRole.MEMBER,
+    add_to_all_projects: bool = False,
+) -> None:
+    """Creates invitation to Neptune workspace.
+
+    Args:
+        username: username of the user to invite.
+        email: email of the user to invite.
+            Note: at least one of the above parameters are needed.
+            If neither the username nor the email is passed, will raise ValueError.
+            If both are filled, will raise ValueError.
+        workspace: Name of your Neptune workspace.
+        api_token: Account's API token.
+            If None, the value of the NEPTUNE_API_TOKEN environment variable is used.
+            Note: To keep your token secure, use the NEPTUNE_API_TOKEN environment variable rather than placing your
+            API token in plain text in your source code.
+        role: The workspace role that is to be granted to the invited user.
+        You can choose between the following values:
+        - Administrator: `WorkspaceMemberRole.ADMIN`
+        - Member: `WorkspaceMemberRole.MEMBER`
+        add_to_all_projects: Whether to add the user to all projects in the workspace.
+
+    Example:
+        >>> from neptune import management
+        >>> from management import WorkspaceMemberRole
+        >>> management.invite_to_workspace(
+        ...     username="user",
+        ...     workspace="ml-team",
+        ...     role=WorkspaceMemberRole.ADMIN,
+        ... )
+
+    You may also want to check the management API reference:
+    https://docs.neptune.ai/api/management
+    """
+    verify_type("workspace", workspace, str)
+    verify_type("role", role, (WorkspaceMemberRole, str))
+    verify_type("add_to_all_projects", add_to_all_projects, bool)
+    verify_type("username", username, (str, type(None)))
+    verify_type("email", email, (str, type(None)))
+    verify_type("api_token", api_token, (str, type(None)))
+
+    if username and email:
+        raise ValueError("Cannot specify both `username` and `email`.")
+
+    if username:
+        invitee = username
+        invitation_type = "user"
+    elif email:
+        invitee = email
+        invitation_type = "emailRecipient"
+    else:
+        raise ValueError("Neither `username` nor `email` arguments filled. At least one needs to be passed")
+
+    if isinstance(role, str):
+        if role.lower() not in {"admin", "owner", "member"}:
+            raise ValueError(f"Unrecognized role: {role}")
+
+        role = role.lower()
+
+        role = role if role != "admin" else "owner"
+
+    if isinstance(role, WorkspaceMemberRole):
+        role = role.value
+
+    params = {
+        "newOrganizationInvitations": {
+            "invitationsEntries": [
+                {
+                    "invitee": invitee,
+                    "invitationType": invitation_type,
+                    "roleGrant": role,
+                    "addToAllProjects": add_to_all_projects,
+                }
+            ],
+            "organizationIdentifier": workspace,
+        },
+        **DEFAULT_REQUEST_KWARGS,
+    }
+
+    backend_client = _get_backend_client(api_token=api_token)
+    try:
+        backend_client.api.createOrganizationInvitations(**params)
+    except HTTPNotFound:
+        raise WorkspaceOrUserNotFound(workspace=workspace, user=invitee)
+    except HTTPConflict:
+        raise UserAlreadyInvited(user=invitee, workspace=workspace)
+
+
 @with_api_exceptions_handler
 def get_project_service_account_list(
     project: str, *, workspace: Optional[str] = None, api_token: Optional[str] = None

src/neptune/management/internal/utils.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 import re
+from enum import Enum
 from typing import Optional
 
 from neptune.common.patterns import PROJECT_QUALIFIED_NAME_PATTERN
@@ -50,3 +51,8 @@ def normalize_project_name(name: str, workspace: Optional[str] = None):
     extracted_workspace_name, extracted_project_name = extract_project_and_workspace(name=name, workspace=workspace)
 
     return f"{extracted_workspace_name}/{extracted_project_name}"
+
+
+class WorkspaceMemberRole(Enum):
+    MEMBER = "member"
+    ADMIN = "owner"

tests/e2e/management/test_management.py

@@ -30,11 +30,16 @@
     get_project_service_account_list,
     get_workspace_member_list,
     get_workspace_service_account_list,
+    invite_to_workspace,
     remove_project_member,
     remove_project_service_account,
     trash_objects,
 )
-from neptune.management.exceptions import UserNotExistsOrWithoutAccess
+from neptune.management.exceptions import (
+    UserAlreadyInvited,
+    UserNotExistsOrWithoutAccess,
+    WorkspaceOrUserNotFound,
+)
 from neptune.management.internal.utils import normalize_project_name
 from tests.e2e.base import (
     BaseE2ETest,
@@ -348,6 +353,22 @@ def test_invite_as_non_admin(self, environment: "Environment"):
 
         assert project_identifier not in get_project_list(api_token=environment.user_token)
 
+    def test_invite_to_workspace(self, environment: "Environment"):
+        with pytest.raises(UserAlreadyInvited):
+            invite_to_workspace(
+                username=environment.user, workspace=environment.workspace, api_token=environment.admin_token
+            )
+
+        with pytest.raises(WorkspaceOrUserNotFound):
+            invite_to_workspace(
+                username="non-existent-user", workspace=environment.workspace, api_token=environment.admin_token
+            )
+
+        with pytest.raises(WorkspaceOrUserNotFound):
+            invite_to_workspace(
+                username=environment.user, workspace="non-existent-workspace", api_token=environment.admin_token
+            )
+
 
 @pytest.mark.management
 class TestTrashObjects(BaseE2ETest):

tests/unit/neptune/new/internal/backends/test_hosted_client.py

@@ -31,6 +31,7 @@
 )
 
 from neptune.internal.backends.hosted_client import (
+    DEFAULT_REQUEST_KWARGS,
     _get_token_client,
     create_artifacts_client,
     create_backend_client,
@@ -47,6 +48,7 @@
     get_project_list,
     get_project_member_list,
     get_workspace_member_list,
+    invite_to_workspace,
     remove_project_member,
 )
 from neptune.management.exceptions import (
@@ -115,6 +117,33 @@ def test_project_listing(self, swagger_client_factory):
         # then:
         self.assertEqual(["org1/project1", "org2/project2"], returned_projects)
 
+    def test_invite_to_workspace(self, swagger_client_factory):
+        # given:
+        swagger_client = self._get_swagger_client_mock(swagger_client_factory)
+
+        # when:
+        invite_to_workspace(
+            username="tester1",
+            workspace="org2",
+            api_token=API_TOKEN,
+        )
+
+        # then:
+        self.assertEqual(swagger_client.api.createOrganizationInvitations.call_count, 1)
+
+        swagger_client.api.createOrganizationInvitations.assert_called_once_with(
+            newOrganizationInvitations={
+                "invitationsEntries": [
+                    {"invitee": "tester1", "invitationType": "user", "roleGrant": "member", "addToAllProjects": False}
+                ],
+                "organizationIdentifier": "org2",
+            },
+            **DEFAULT_REQUEST_KWARGS,
+        )
+
+    def test_invite_to_workspace_no_username_email_raises(self, swagger_client_factory):
+        self.assertRaises(ValueError, invite_to_workspace, workspace="org2", api_token=API_TOKEN)
+
     def test_workspace_members(self, swagger_client_factory):
         swagger_client = self._get_swagger_client_mock(swagger_client_factory)