Get invitations

libs/labelbox/src/labelbox/schema/invite.py

@@ -1,9 +1,15 @@
+from typing import TYPE_CHECKING
 from dataclasses import dataclass
 
 from labelbox.orm.db_object import DbObject
 from labelbox.orm.model import Field
 from labelbox.schema.role import ProjectRole, format_role
 
+from labelbox.pagination import PaginatedCollection
+
+if TYPE_CHECKING:
+    from labelbox import Client
+
 
 @dataclass
 class InviteLimit:
@@ -31,10 +37,138 @@ def __init__(self, client, invite_response):
         project_roles = invite_response.pop("projectInvites", [])
         super().__init__(client, invite_response)
 
-        self.project_roles = [
-            ProjectRole(
-                project=client.get_project(r["projectId"]),
-                role=client.get_roles()[format_role(r["projectRoleName"])],
-            )
-            for r in project_roles
-        ]
+        self.project_roles = []
+
+        # If a project is deleted then it doesn't show up in the invite
+        for pr in project_roles:
+            try:
+                project = client.get_project(pr["projectId"])
+                if project:  # Check if project exists
+                    self.project_roles.append(
+                        ProjectRole(
+                            project=project,
+                            role=client.get_roles()[
+                                format_role(pr["projectRoleName"])
+                            ],
+                        )
+                    )
+            except Exception:
+                # Skip this project role if the project is no longer available
+                continue
+
+    def cancel(self) -> bool:
+        """
+        Cancels this invite.
+
+        This will prevent the invited user from accepting the invitation.
+
+        Returns:
+            bool: True if the invite was successfully canceled, False otherwise.
+        """
+
+        # Case of a newly invited user
+        if self.uid == "invited":
+            return False
+
+        query_str = """
+            mutation CancelInvitePyApi($where: WhereUniqueIdInput!) {
+                cancelInvite(where: $where) {
+                    id
+                }
+            }"""
+        result = self.client.execute(
+            query_str, {"where": {"id": self.uid}}, experimental=True
+        )
+        return (
+            result is not None
+            and "cancelInvite" in result
+            and result.get("cancelInvite") is not None
+        )
+
+    @staticmethod
+    def get_project_invites(
+        client: "Client", project_id: str
+    ) -> PaginatedCollection:
+        """
+        Retrieves all invites for a specific project.
+
+        Args:
+            client (Client): The Labelbox client instance.
+            project_id (str): The ID of the project to get invites for.
+
+        Returns:
+            PaginatedCollection: A collection of Invite objects for the specified project.
+        """
+        query = """query GetProjectInvitationsPyApi(
+                    $from: ID
+                    $first: PageSize
+                    $projectId: ID!
+                    ) {
+                    project(where: { id: $projectId }) {
+                        id
+                        invites(from: $from, first: $first) {
+                        nodes {
+                            id
+                            createdAt
+                            organizationRoleName
+                            inviteeEmail
+                            projectInvites {
+                            id
+                            projectRoleName
+                            projectId
+                            }
+                        }
+                        nextCursor
+                        }
+                    }
+                    }"""
+
+        invites = PaginatedCollection(
+            client,
+            query,
+            {"projectId": project_id, "search": ""},
+            ["project", "invites", "nodes"],
+            Invite,
+            cursor_path=["project", "invites", "nextCursor"],
+        )
+        return invites
+
+    @staticmethod
+    def get_invites(client: "Client") -> PaginatedCollection:
+        """
+        Retrieves all invites for the organization.
+
+        Args:
+            client (Client): The Labelbox client instance.
+
+        Returns:
+            PaginatedCollection: A collection of Invite objects for the organization.
+        """
+        query_str = """query GetOrgInvitationsPyApi($from: ID, $first: PageSize) {
+            organization {
+                id
+                invites(from: $from, first: $first) {
+                nodes {
+                    id
+                    createdAt
+                    organizationRoleName
+                    inviteeEmail
+                    projectInvites {
+                    id
+                    projectRoleName
+                    projectId
+                    }
+                }
+                nextCursor
+                }
+            }
+            }"""
+        invites = PaginatedCollection(
+            client,
+            query_str,
+            {},
+            ["organization", "invites", "nodes"],
+            Invite,
+            cursor_path=["organization", "invites", "nextCursor"],
+        )
+        return invites

libs/labelbox/src/labelbox/schema/organization.py

@@ -7,6 +7,7 @@
 from labelbox.orm.model import Field, Relationship
 from labelbox.schema.invite import InviteLimit
 from labelbox.schema.resource_tag import ResourceTag
+from labelbox.pagination import PaginatedCollection
 
 if TYPE_CHECKING:
     from labelbox import (
@@ -243,3 +244,24 @@ def get_default_iam_integration(self) -> Optional["IAMIntegration"]:
         return (
             None if not len(default_integration) else default_integration.pop()
         )
+
+    def get_invites(self) -> PaginatedCollection:
+        """
+        Retrieves all invites for this organization.
+
+        Returns:
+            PaginatedCollection: A collection of Invite objects for the organization.
+        """
+        return Entity.Invite.get_invites(self.client)
+
+    def get_project_invites(self, project_id: str) -> PaginatedCollection:
+        """
+        Retrieves all invites for a specific project in this organization.
+
+        Args:
+            project_id (str): The ID of the project to get invites for.
+
+        Returns:
+            PaginatedCollection: A collection of Invite objects for the specified project.
+        """
+        return Entity.Invite.get_project_invites(self.client, project_id)

libs/labelbox/tests/integration/test_invite.py

@@ -0,0 +1,197 @@
+import pytest
+from faker import Faker
+from labelbox.schema.media_type import MediaType
+from labelbox import ProjectRole
+import time
+
+faker = Faker()
+
+
+@pytest.fixture
+def dummy_email():
+    """Generate a random dummy email for testing"""
+    return f"none+{faker.uuid4()}@labelbox.com"
+
+
+@pytest.fixture(scope="module")
+def test_project(client):
+    """Create a temporary project for testing"""
+    project = client.create_project(
+        name=f"test-project-{faker.uuid4()}", media_type=MediaType.Image
+    )
+    yield project
+
+    # Occurs after the test is finished based on scope
+    project.delete()
+
+
+@pytest.fixture
+def org_invite(client, dummy_email):
+    """Create an organization-level invite"""
+    role = client.get_roles()["LABELER"]
+    organization = client.get_organization()
+    invite = organization.invite_user(dummy_email, role)
+
+    yield invite
+
+    if invite.uid:
+        invite.cancel()
+
+
+@pytest.fixture
+def project_invite(client, test_project, dummy_email):
+    """Create a project-level invite"""
+    roles = client.get_roles()
+    project_role = ProjectRole(project=test_project, role=roles["LABELER"])
+    organization = client.get_organization()
+
+    invite = organization.invite_user(
+        dummy_email, roles["NONE"], project_roles=[project_role]
+    )
+
+    yield invite
+
+    # Cleanup: Use invite.cancel() instead of organization.cancel_invite()
+    if invite.uid:
+        invite.cancel()
+
+
+def test_get_organization_invites(client, org_invite):
+    """Test retrieving all organization invites"""
+    # Add a small delay to ensure invite is created
+    time.sleep(1)
+
+    organization = client.get_organization()
+    invites = organization.get_invites()
+    invite_list = [invite for invite in invites]
+    assert len(invite_list) > 0
+
+    # Verify our test invite is in the list
+    invite_emails = [invite.email for invite in invite_list]
+    assert org_invite.email in invite_emails
+
+
+def test_get_project_invites(client, test_project, project_invite):
+    """Test retrieving project-specific invites"""
+    # Add a small delay to ensure invite is created
+    time.sleep(1)
+
+    organization = client.get_organization()
+    project_invites = organization.get_project_invites(test_project.uid)
+    invite_list = [invite for invite in project_invites]
+    assert len(invite_list) > 0
+
+    # Verify our test invite is in the list
+    invite_emails = [invite.email for invite in invite_list]
+    assert project_invite.email in invite_emails
+
+    # Verify project role assignment
+    found_invite = next(
+        invite for invite in invite_list if invite.email == project_invite.email
+    )
+    assert len(found_invite.project_roles) == 1
+    assert found_invite.project_roles[0].project.uid == test_project.uid
+
+
+def test_cancel_invite(client, dummy_email):
+    """Test canceling an invite"""
+    # Create a new invite
+    role = client.get_roles()["LABELER"]
+    organization = client.get_organization()
+    organization.invite_user(dummy_email, role)
+
+    # Add a small delay to ensure invite is created
+    time.sleep(1)
+
+    # Find the actual invite by email
+    invites = organization.get_invites()
+    found_invite = next(
+        (invite for invite in invites if invite.email == dummy_email), None
+    )
+    assert found_invite is not None, f"Invite for {dummy_email} not found"
+
+    # Cancel the invite using the found invite object
+    result = found_invite.cancel()
+    assert result is True
+
+    # Verify the invite is no longer in the organization's invites
+    invites = organization.get_invites()
+    invite_emails = [i.email for i in invites]
+    assert dummy_email not in invite_emails
+
+
+def test_cancel_project_invite(client, test_project, dummy_email):
+    """Test canceling a project invite"""
+    # Create a project invite
+    roles = client.get_roles()
+    project_role = ProjectRole(project=test_project, role=roles["LABELER"])
+    organization = client.get_organization()
+
+    organization.invite_user(
+        dummy_email, roles["NONE"], project_roles=[project_role]
+    )
+
+    # Add a small delay to ensure invite is created
+    time.sleep(1)
+
+    # Find the actual invite by email
+    invites = organization.get_invites()
+    found_invite = next(
+        (invite for invite in invites if invite.email == dummy_email), None
+    )
+    assert found_invite is not None, f"Invite for {dummy_email} not found"
+
+    # Cancel the invite using the found invite object
+    result = found_invite.cancel()
+    assert result is True
+
+    # Verify the invite is no longer in the project's invites
+    project_invites = organization.get_project_invites(test_project.uid)
+    invite_emails = [i.email for i in project_invites]
+    assert dummy_email not in invite_emails
+
+
+def test_project_invite_after_project_deletion(client, dummy_email):
+    """Test that project invites are properly filtered when a project is deleted"""
+    # Create two test projects
+    project1 = client.create_project(
+        name=f"test-project1-{faker.uuid4()}", media_type=MediaType.Image
+    )
+    project2 = client.create_project(
+        name=f"test-project2-{faker.uuid4()}", media_type=MediaType.Image
+    )
+
+    # Create project roles
+    roles = client.get_roles()
+    project_role1 = ProjectRole(project=project1, role=roles["LABELER"])
+    project_role2 = ProjectRole(project=project2, role=roles["LABELER"])
+
+    # Invite user to both projects
+    organization = client.get_organization()
+    organization.invite_user(
+        dummy_email, roles["NONE"], project_roles=[project_role1, project_role2]
+    )
+
+    # Add a small delay to ensure invite is created
+    time.sleep(1)
+
+    # Delete one project
+    project1.delete()
+
+    # Find the invite and verify project roles
+    invites = organization.get_invites()
+    found_invite = next(
+        (invite for invite in invites if invite.email == dummy_email), None
+    )
+    assert found_invite is not None, f"Invite for {dummy_email} not found"
+
+    # Verify only one project role remains
+    assert (
+        len(found_invite.project_roles) == 1
+    ), "Expected only one project role"
+    assert found_invite.project_roles[0].project.uid == project2.uid
+
+    # Cleanup
+    project2.delete()
+    if found_invite.uid:
+        found_invite.cancel()