Add users invite batch support (#323)

Author: dorsha

.gitignore

@@ -130,3 +130,5 @@ dmypy.json
 
 # Mac OS
 .DS_Store
+
+.history

README.md

@@ -557,6 +557,24 @@ descope_client.mgmt.user.invite(
     ],
 )
 
+# Batch invite
+descope_client.mgmt.user.invite_batch(
+    users=[
+        UserObj(
+            login_id="[email protected]",
+            email="[email protected]",
+            display_name="Desmond Copeland",
+            user_tenants=[
+                AssociatedTenant("my-tenant-id", ["role-name1"]),
+            ],
+            custom_attributes={"ak": "av"},
+        )
+    ],
+    invite_url="invite.me",
+    send_mail=True,
+    send_sms=True,
+)
+
 # Update will override all fields as is. Use carefully.
 descope_client.mgmt.user.update(
     login_id="[email protected]",

descope/management/common.py

@@ -12,6 +12,7 @@ class MgmtV1:
 
     # user
     user_create_path = "/v1/mgmt/user/create"
+    user_create_batch_path = "/v1/mgmt/user/create/batch"
     user_update_path = "/v1/mgmt/user/update"
     user_delete_path = "/v1/mgmt/user/delete"
     user_logout_path = "/v1/mgmt/user/logout"

descope/management/user.py

@@ -11,6 +11,40 @@
 )
 
 
+class UserObj:
+    def __init__(
+        self,
+        login_id: str,
+        email: Optional[str] = None,
+        phone: Optional[str] = None,
+        display_name: Optional[str] = None,
+        given_name: Optional[str] = None,
+        middle_name: Optional[str] = None,
+        family_name: Optional[str] = None,
+        role_names: Optional[List[str]] = None,
+        user_tenants: Optional[List[AssociatedTenant]] = None,
+        picture: Optional[str] = None,
+        custom_attributes: Optional[dict] = None,
+        verified_email: Optional[bool] = None,
+        verified_phone: Optional[bool] = None,
+        additional_login_ids: Optional[List[str]] = None,
+    ):
+        self.login_id = login_id
+        self.email = email
+        self.phone = phone
+        self.display_name = display_name
+        self.given_name = given_name
+        self.middle_name = middle_name
+        self.family_name = family_name
+        self.role_names = role_names
+        self.user_tenants = user_tenants
+        self.picture = picture
+        self.custom_attributes = custom_attributes
+        self.verified_email = verified_email
+        self.verified_phone = verified_phone
+        self.additional_login_ids = additional_login_ids
+
+
 class User(AuthBase):
     def create(
         self,
@@ -181,13 +215,13 @@ def invite(
         additional_login_ids: Optional[List[str]] = None,
     ) -> dict:
         """
-        Create a new user and invite them via an email message.
+        Create a new user and invite them via an email / text message.
 
         Functions exactly the same as the `create` function with the additional invitation
             behavior. See the documentation above for the general creation behavior.
 
-        IMPORTANT: Since the invitation is sent by email, make sure either
-            the email is explicitly set, or the login_id itself is an email address.
+        IMPORTANT: Since the invitation is sent by email / phone, make sure either
+            the email / phone is explicitly set, or the login_id itself is an email address / phone number.
             You must configure the invitation URL in the Descope console prior to
             calling the method.
         """
@@ -221,6 +255,41 @@ def invite(
         )
         return response.json()
 
+    def invite_batch(
+        self,
+        users: List[UserObj],
+        invite_url: Optional[str] = None,
+        send_mail: Optional[
+            bool
+        ] = None,  # send invite via mail, default is according to project settings
+        send_sms: Optional[
+            bool
+        ] = None,  # send invite via text message, default is according to project settings
+    ) -> dict:
+        """
+        Create users in batch and invite them via an email / text message.
+
+        Functions exactly the same as the `create` function with the additional invitation
+            behavior. See the documentation above for the general creation behavior.
+
+        IMPORTANT: Since the invitation is sent by email / phone, make sure either
+            the email / phone is explicitly set, or the login_id itself is an email address / phone number.
+            You must configure the invitation URL in the Descope console prior to
+            calling the method.
+        """
+
+        response = self._auth.do_post(
+            MgmtV1.user_create_batch_path,
+            User._compose_create_batch_body(
+                users,
+                invite_url,
+                send_mail,
+                send_sms,
+            ),
+            pswd=self._auth.management_key,
+        )
+        return response.json()
+
     def update(
         self,
         login_id: str,
@@ -1193,6 +1262,45 @@ def _compose_create_body(
             body["sendSMS"] = send_sms
         return body
 
+    @staticmethod
+    def _compose_create_batch_body(
+        users: List[UserObj],
+        invite_url: Optional[str],
+        send_mail: Optional[bool],
+        send_sms: Optional[bool],
+    ) -> dict:
+        usersBody = []
+        for user in users:
+            role_names = [] if user.role_names is None else user.role_names
+            user_tenants = [] if user.user_tenants is None else user.user_tenants
+            uBody = User._compose_update_body(
+                login_id=user.login_id,
+                email=user.email,
+                phone=user.phone,
+                display_name=user.display_name,
+                given_name=user.given_name,
+                middle_name=user.middle_name,
+                family_name=user.family_name,
+                role_names=role_names,
+                user_tenants=user_tenants,
+                picture=user.picture,
+                custom_attributes=user.custom_attributes,
+                additional_login_ids=user.additional_login_ids,
+                verified_email=user.verified_email,
+                verified_phone=user.verified_phone,
+                test=False,
+            )
+            usersBody.append(uBody)
+
+        body = {"users": usersBody, "invite": True}
+        if invite_url is not None:
+            body["inviteUrl"] = invite_url
+        if send_mail is not None:
+            body["sendMail"] = send_mail
+        if send_sms is not None:
+            body["sendSMS"] = send_sms
+        return body
+
     @staticmethod
     def _compose_update_body(
         login_id: str,

tests/management/test_user.py

@@ -5,6 +5,7 @@
 from descope import AssociatedTenant, AuthException, DescopeClient
 from descope.common import DEFAULT_TIMEOUT_SECONDS, DeliveryMethod, LoginOptions
 from descope.management.common import MgmtV1
+from descope.management.user import UserObj
 
 from .. import common
 
@@ -256,6 +257,77 @@ def test_invite(self):
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
 
+    def test_invite_batch(self):
+        # Test failed flows
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = False
+            self.assertRaises(
+                AuthException,
+                self.client.mgmt.user.invite_batch,
+                [],
+            )
+
+        # Test success flow
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads("""{"users": [{"id": "u1"}]}""")
+            mock_post.return_value = network_resp
+            resp = self.client.mgmt.user.invite_batch(
+                users=[
+                    UserObj(
+                        login_id="[email protected]",
+                        email="[email protected]",
+                        display_name="Name",
+                        user_tenants=[
+                            AssociatedTenant("tenant1"),
+                            AssociatedTenant("tenant2", ["role1", "role2"]),
+                        ],
+                        custom_attributes={"ak": "av"},
+                    )
+                ],
+                invite_url="invite.me",
+                send_sms=True,
+            )
+            users = resp["users"]
+            self.assertEqual(users[0]["id"], "u1")
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.user_create_batch_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                },
+                params=None,
+                json={
+                    "users": [
+                        {
+                            "loginId": "[email protected]",
+                            "email": "[email protected]",
+                            "phone": None,
+                            "displayName": "Name",
+                            "roleNames": [],
+                            "userTenants": [
+                                {"tenantId": "tenant1", "roleNames": []},
+                                {
+                                    "tenantId": "tenant2",
+                                    "roleNames": ["role1", "role2"],
+                                },
+                            ],
+                            "test": False,
+                            "picture": None,
+                            "customAttributes": {"ak": "av"},
+                            "additionalLoginIds": None,
+                        }
+                    ],
+                    "invite": True,
+                    "inviteUrl": "invite.me",
+                    "sendSMS": True,
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
     def test_update(self):
         # Test failed flows
         with patch("requests.post") as mock_post: