Add invitations API methods

tests/test_user_management.py

@@ -1,9 +1,10 @@
 import pytest
 
-from tests.utils.fixtures.mock_session import MockSession
-from tests.utils.fixtures.mock_user import MockUser
 from tests.utils.fixtures.mock_auth_factor_totp import MockAuthFactorTotp
+from tests.utils.fixtures.mock_invitation import MockInvitation
 from tests.utils.fixtures.mock_organization_membership import MockOrganizationMembership
+from tests.utils.fixtures.mock_session import MockSession
+from tests.utils.fixtures.mock_user import MockUser
 from workos.user_management import UserManagement
 
 
@@ -193,6 +194,32 @@ def mock_auth_factors(self):
         }
         return dict_response
 
+    @pytest.fixture
+    def mock_invitation(self):
+        return MockInvitation("invitation_ABCDE").to_dict()
+
+    @pytest.fixture
+    def mock_invitations(self):
+        invitation_list = [MockInvitation(id=str(i)).to_dict() for i in range(50)]
+
+        dict_response = {
+            "data": invitation_list,
+            "list_metadata": {"before": None, "after": None},
+            "metadata": {
+                "params": {
+                    "email": None,
+                    "organization_id": None,
+                    "limit": None,
+                    "before": None,
+                    "after": None,
+                    "order": None,
+                    "default_limit": True,
+                },
+                "method": UserManagement.list_invitations,
+            },
+        }
+        return dict_response
+
     def test_create_user(self, mock_user, mock_request_method):
         mock_request_method("post", mock_user, 201)
 
@@ -525,3 +552,45 @@ def test_auth_factors_returns_metadata(
 
         dict_auth_factors = auth_factors.to_dict()
         assert dict_auth_factors["metadata"]["params"]["user_id"] == "user_12345"
+
+    def test_get_invitation(self, mock_invitation, capture_and_mock_request):
+        url, request_kwargs = capture_and_mock_request("get", mock_invitation, 200)
+
+        invitation = self.user_management.get_invitation("invitation_ABCDE")
+
+        assert url[0].endswith("user_management/invitations/invitation_ABCDE")
+        assert invitation["id"] == "invitation_ABCDE"
+
+    def test_list_invitations_returns_metadata(
+        self,
+        mock_invitations,
+        mock_request_method,
+    ):
+        mock_request_method("get", mock_invitations, 200)
+
+        invitations = self.user_management.list_invitations(
+            organization_id="org_12345",
+        )
+
+        dict_invitations = invitations.to_dict()
+        assert dict_invitations["metadata"]["params"]["organization_id"] == "org_12345"
+
+    def test_send_invitation(self, capture_and_mock_request, mock_invitation):
+        email = "[email protected]"
+        organization_id = "org_12345"
+        url, _ = capture_and_mock_request("post", mock_invitation, 201)
+
+        invitation = self.user_management.send_invitation(
+            email=email, organization_id=organization_id
+        )
+
+        assert url[0].endswith("user_management/invitations")
+        assert invitation["email"] == email
+        assert invitation["organization_id"] == organization_id
+
+    def test_revoke_invitation(self, capture_and_mock_request, mock_invitation):
+        url, _ = capture_and_mock_request("post", mock_invitation, 200)
+
+        user = self.user_management.revoke_invitation("invitation_ABCDE")
+
+        assert url[0].endswith("user_management/invitations/invitation_ABCDE/revoke")

tests/utils/fixtures/mock_invitation.py

@@ -0,0 +1,29 @@
+import datetime
+from workos.resources.base import WorkOSBaseResource
+
+
+class MockInvitation(WorkOSBaseResource):
+    def __init__(self, id):
+        self.id = id
+        self.email = "[email protected]"
+        self.state = "pending"
+        self.accepted_at = None
+        self.revoked_at = None
+        self.expires_at = datetime.datetime.now()
+        self.token = "ABCDE12345"
+        self.organization_id = "org_12345"
+        self.created_at = datetime.datetime.now()
+        self.updated_at = datetime.datetime.now()
+
+    OBJECT_FIELDS = [
+        "id",
+        "email",
+        "state",
+        "accepted_at",
+        "revoked_at",
+        "expires_at",
+        "token",
+        "organization_id",
+        "created_at",
+        "updated_at",
+    ]

workos/resources/user_management.py

@@ -26,6 +26,27 @@ def to_dict(self):
         return authentication_response_dict
 
 
+class WorkOSInvitation(WorkOSBaseResource):
+    """Representation of an Invitation as returned by WorkOS through User Management features.
+
+    Attributes:
+        OBJECT_FIELDS (list): List of fields a WorkOSInvitation comprises.
+    """
+
+    OBJECT_FIELDS = [
+        "id",
+        "email",
+        "state",
+        "accepted_at",
+        "revoked_at",
+        "expires_at",
+        "token",
+        "organization_id",
+        "created_at",
+        "updated_at",
+    ]
+
+
 class WorkOSOrganizationMembership(WorkOSBaseResource):
     """Representation of an Organization Membership as returned by WorkOS through User Management features.
 

workos/user_management.py

@@ -3,6 +3,7 @@
 from workos.resources.mfa import WorkOSAuthenticationFactorTotp, WorkOSChallenge
 from workos.resources.user_management import (
     WorkOSAuthenticationResponse,
+    WorkOSInvitation,
     WorkOSOrganizationMembership,
     WorkOSPasswordChallengeResponse,
     WorkOSUser,
@@ -29,6 +30,9 @@
 USER_VERIFY_EMAIL_CODE_PATH = "users/verify_email_code"
 USER_SEND_MAGIC_AUTH_PATH = "user_management/magic_auth/send"
 USER_AUTH_FACTORS_PATH = "user_management/users/{0}/auth_factors"
+INVITATION_PATH = "user_management/invitations"
+INVITATION_DETAIL_PATH = "user_management/invitations/{0}"
+INVITATION_REVOKE_PATH = "user_management/invitations/{0}/revoke"
 
 RESPONSE_LIMIT = 10
 
@@ -698,3 +702,141 @@ def list_auth_factors(
         }
 
         return self.construct_from_response(response)
+
+    def get_invitation(self, invitation_id):
+        """Get the details of an invitation.
+
+        Args:
+            invitation_id (str) -  The unique ID of the Invitation.
+
+        Returns:
+            dict: Invitation response from WorkOS.
+        """
+        headers = {}
+
+        response = self.request_helper.request(
+            INVITATION_DETAIL_PATH.format(invitation_id),
+            method=REQUEST_METHOD_GET,
+            headers=headers,
+            token=workos.api_key,
+        )
+
+        return WorkOSInvitation.construct_from_response(response).to_dict()
+
+    def list_invitations(
+        self,
+        email=None,
+        organization_id=None,
+        limit=None,
+        before=None,
+        after=None,
+        order=None,
+    ):
+        """Get a list of all of your existing invitations matching the criteria specified.
+
+        Kwargs:
+            email (str): Filter Invitations by email. (Optional)
+            organization_id (str): Filter Invitations by organization. (Optional)
+            limit (int): Maximum number of records to return. (Optional)
+            before (str): Pagination cursor to receive records before a provided Invitation ID. (Optional)
+            after (str): Pagination cursor to receive records after a provided Invitation ID. (Optional)
+            order (Order): Sort records in either ascending or descending order by created_at timestamp: "asc" or "desc" (Optional)
+
+        Returns:
+            dict: Users response from WorkOS.
+        """
+
+        default_limit = None
+
+        if limit is None:
+            limit = RESPONSE_LIMIT
+            default_limit = True
+
+        params = {
+            "email": email,
+            "organization_id": organization_id,
+            "limit": limit,
+            "before": before,
+            "after": after,
+        }
+
+        if order is not None:
+            if isinstance(order, Order):
+                params["order"] = str(order.value)
+            elif order == "asc" or order == "desc":
+                params["order"] = order
+            else:
+                raise ValueError("Parameter order must be of enum type Order")
+
+        response = self.request_helper.request(
+            INVITATION_PATH,
+            method=REQUEST_METHOD_GET,
+            params=params,
+            token=workos.api_key,
+        )
+
+        response["metadata"] = {
+            "params": params,
+            "method": UserManagement.list_invitations,
+        }
+
+        if "default_limit" in locals():
+            if "metadata" in response and "params" in response["metadata"]:
+                response["metadata"]["params"]["default_limit"] = default_limit
+            else:
+                response["metadata"] = {"params": {"default_limit": default_limit}}
+
+        return self.construct_from_response(response)
+
+    def send_invitation(
+        self, email, organization_id=None, expires_in_days=None, inviter_user_id=None
+    ):
+        """Sends an Invitation to a recipient.
+
+        Args:
+            email: The email address of the recipient.
+            organization_id: The ID of the Organization to which the recipient is being invited. (Optional)
+            expires_in_days: The number of days the invitations will be valid for. Must be between 1 and 30, defaults to 7 if not specified. (Optional)
+            inviter_user_id: The ID of the User sending the invitation. (Optional)
+
+        Returns:
+            dict: Sent Invitation response from WorkOS.
+        """
+        headers = {}
+
+        params = {
+            "email": email,
+            "organization_id": organization_id,
+            "expires_in_days": expires_in_days,
+            "inviter_user_id": inviter_user_id,
+        }
+
+        response = self.request_helper.request(
+            INVITATION_PATH,
+            method=REQUEST_METHOD_POST,
+            params=params,
+            headers=headers,
+            token=workos.api_key,
+        )
+
+        return WorkOSInvitation.construct_from_response(response).to_dict()
+
+    def revoke_invitation(self, invitation_id):
+        """Revokes an existing Invitation.
+
+        Args:
+            invitation_id (str) -  The unique ID of the Invitation.
+
+        Returns:
+            dict: Invitation response from WorkOS.
+        """
+        headers = {}
+
+        response = self.request_helper.request(
+            INVITATION_REVOKE_PATH.format(invitation_id),
+            method=REQUEST_METHOD_POST,
+            headers=headers,
+            token=workos.api_key,
+        )
+
+        return WorkOSInvitation.construct_from_response(response).to_dict()