Set project owner during project creation

Author: midhun-pm

libs/labelbox/src/labelbox/alignerr/alignerr_project.py

@@ -6,6 +6,7 @@
 from labelbox.alignerr.schema.project_rate import ProjectRateV2
 from labelbox.alignerr.schema.project_domain import ProjectDomain
 from labelbox.alignerr.schema.enchanced_resource_tags import EnhancedResourceTag, ResourceTagType
+from labelbox.alignerr.schema.project_boost_workforce import ProjectBoostWorkforce
 from labelbox.pagination import PaginatedCollection
 
 logger = logging.getLogger(__name__)
@@ -132,6 +133,17 @@ def remove_tag(self, tag: EnhancedResourceTag):
         self.set_tags(current_tag_names)
         return self
 
+    def get_project_owner(self) -> Optional[ProjectBoostWorkforce]:
+        """Get the ProjectBoostWorkforce for this project.
+        
+        Returns:
+            ProjectBoostWorkforce instance or None if not found
+        """
+        return ProjectBoostWorkforce.get_by_project_id(
+            client=self.client, 
+            project_id=self.project.uid
+        )
+
 
 class AlignerrWorkspace:
     def __init__(self, client: "Client"):

libs/labelbox/src/labelbox/alignerr/alignerr_project_builder.py

@@ -1,16 +1,25 @@
 import datetime
-from typing import TYPE_CHECKING, Optional
+from enum import Enum
+from typing import TYPE_CHECKING, Optional, Union, List
 import logging
 
 from labelbox.alignerr.schema.project_rate import BillingMode
 from labelbox.alignerr.schema.project_rate import ProjectRateInput
 from labelbox.alignerr.schema.project_domain import ProjectDomain
 from labelbox.alignerr.schema.enchanced_resource_tags import EnhancedResourceTag, ResourceTagType
+from labelbox.alignerr.schema.project_boost_workforce import ProjectBoostWorkforce
 from labelbox.schema.media_type import MediaType
 
 logger = logging.getLogger(__name__)
 
 
+class ValidationType(Enum):
+    """Enum for validation types that can be selectively skipped."""
+    ALIGNERR_RATE = "AlignerrRate"
+    CUSTOMER_RATE = "CustomerRate"
+    PROJECT_OWNER = "ProjectOwner"
+
+
 if TYPE_CHECKING:
     from labelbox import Client
     from labelbox.alignerr.alignerr_project import AlignerrProject, AlignerrRole
@@ -23,6 +32,7 @@ def __init__(self, client: "Client"):
         self._customer_rate: ProjectRateInput = None
         self._domains: list[ProjectDomain] = []
         self._enhanced_resource_tags: list[EnhancedResourceTag] = []
+        self._project_owner_email: Optional[str] = None
         self.role_name_to_id = self._get_role_name_to_id()
 
     def set_name(self, name: str):
@@ -141,10 +151,24 @@ def set_tags(self, tag_texts: list[str], tag_type: ResourceTagType):
                 self._enhanced_resource_tags.append(new_tag)
         return self
 
+    def set_project_owner(self, project_owner_email: str):
+        """Set the project owner for the ProjectBoostWorkforce.
+        
+        Args:
+            project_owner_email: Email of the user to set as project owner
+            
+        Returns:
+            Self for method chaining
+        """
+        self._project_owner_email = project_owner_email
+        return self
 
-    def create(self, skip_validation: bool = False):
+
+    def create(self, skip_validation: Union[bool, List[ValidationType]] = False):
         if not skip_validation:
             self._validate()
+        elif isinstance(skip_validation, list):
+            self._validate_selective(skip_validation)
         logger.info("Creating project")
 
         project_data = {
@@ -163,6 +187,7 @@ def create(self, skip_validation: bool = False):
         self._create_rates(alignerr_project)
         self._create_domains(alignerr_project)
         self._create_resource_tags(alignerr_project)
+        self._create_project_owner(alignerr_project)
 
         return alignerr_project
 
@@ -202,6 +227,22 @@ def _create_resource_tags(self, alignerr_project: "AlignerrProject"):
                 tag_type_enum = ResourceTagType(tag_type_str)
                 alignerr_project.set_tags(tag_names, tag_type_enum)
 
+    def _create_project_owner(self, alignerr_project: "AlignerrProject"):
+        if self._project_owner_email:
+            logger.info(f"Setting project owner: {self._project_owner_email}")
+            
+            # Find user by email in the organization
+            user_id = self._find_user_by_email(self._project_owner_email)
+            if not user_id:
+                current_org = self.client.get_organization()
+                raise ValueError(f"User with email {self._project_owner_email} not found in organization {current_org.uid}")
+            
+            ProjectBoostWorkforce.set_project_owner(
+                client=self.client,
+                project_id=alignerr_project.project.uid,
+                project_owner_user_id=user_id
+            )
+
     def _validate_alignerr_rates(self):
         # Import here to avoid circular imports
         from labelbox.alignerr.alignerr_project import AlignerrRole
@@ -221,10 +262,61 @@ def _validate_customer_rate(self):
         if self._customer_rate is None:
             raise ValueError("Customer rate is not set")
 
+    def _validate_project_owner(self):
+        if self._project_owner_email is None:
+            raise ValueError("Project owner is not set")
+
     def _validate(self):
         self._validate_alignerr_rates()
         self._validate_customer_rate()
+        self._validate_project_owner()
+
+    def _validate_selective(self, skip_validations: List[ValidationType]):
+        """Run validations selectively, skipping those in the provided list.
+        
+        Args:
+            skip_validations: List of ValidationType enums to skip
+        """
+        if ValidationType.ALIGNERR_RATE not in skip_validations:
+            self._validate_alignerr_rates()
+        
+        if ValidationType.CUSTOMER_RATE not in skip_validations:
+            self._validate_customer_rate()
+        
+        if ValidationType.PROJECT_OWNER not in skip_validations:
+            self._validate_project_owner()
 
     def _get_role_name_to_id(self) -> dict[str, str]:
         roles = self.client.get_roles()
         return {role.name: role.uid for role in roles.values()}
+
+    def _find_user_by_email(self, email: str) -> Optional[str]:
+        """Find user ID by email in the organization.
+        
+        Args:
+            email: Email address to search for
+            
+        Returns:
+            User ID if found, None otherwise
+        """
+        try:
+            # Import here to avoid circular imports
+            from labelbox.schema.user import User
+            
+            # Get the current organization
+            current_org = self.client.get_organization()
+            
+            # Use client.get_users with where clause to find user by email
+            users = self.client.get_users(where=User.email == email)
+            
+            # Get the first matching user and verify they belong to the same organization
+            user = next(users, None)
+            if user and user.organization().uid == current_org.uid:
+                return user.uid
+            else:
+                logger.warning(f"User with email {email} not found in organization {current_org.uid}")
+                return None
+                    
+        except Exception as e:
+            logger.error(f"Error finding user by email {email}: {e}")
+            return None