feat(sdk): add job-attachments ease-of-use methods

Author: alexenica

src/uipath/_services/__init__.py

@@ -6,6 +6,7 @@
 from .connections_service import ConnectionsService
 from .context_grounding_service import ContextGroundingService
 from .folder_service import FolderService
+from .job_attachments_service import JobAttachmentsService
 from .jobs_service import JobsService
 from .llm_gateway_service import UiPathLlmChatService, UiPathOpenAIService
 from .processes_service import ProcessesService
@@ -22,6 +23,7 @@
     "ApiClient",
     "QueuesService",
     "JobsService",
+    "JobAttachmentsService",
     "UiPathOpenAIService",
     "UiPathLlmChatService",
     "FolderService",

src/uipath/_services/job_attachments_service.py

@@ -0,0 +1,121 @@
+import os
+import shutil
+import tempfile
+import uuid
+from typing import Optional, Union
+
+from .._config import Config
+from .._execution_context import ExecutionContext
+from .._folder_context import FolderContext
+from ._base_service import BaseService
+from .attachments_service import AttachmentsService
+from .jobs_service import JobsService
+
+
+class JobAttachmentsService(FolderContext, BaseService):
+    """Service for managing job attachments in UiPath Orchestrator, focusing on operations related to the current job context.
+
+    Provides methods to attach files to the current job (uploading and linking), and download attachments from the current job.
+    Delegates direct attachment upload/download to AttachmentsService and linking to JobsService.
+    Reference: https://docs.uipath.com/orchestrator/reference/api-job-attachments (Note: This service no longer lists attachments directly)
+    """
+
+    def __init__(self, config: Config, execution_context: ExecutionContext) -> None:
+        super().__init__(config=config, execution_context=execution_context)
+        self._attachments_service = AttachmentsService(config, execution_context)
+        self._jobs_service = JobsService(config, execution_context)
+
+    def attach_file_to_current_job(
+        self,
+        *,
+        name: str,
+        content: Optional[Union[str, bytes]] = None,
+        source_path: Optional[str] = None,
+        category: Optional[str] = None,
+    ) -> Union[uuid.UUID, str]:
+        """Attach a file to the current job, or write to temp if no job context.
+
+        If a job key is present in the execution context, uploads the payload (from content or source_path) using AttachmentsService and links it to the current job using JobsService. Otherwise, writes/copies the file to the system temp directory.
+
+        Args:
+            name (str): The name of the file/attachment.
+            content (Optional[Union[str, bytes]]): The file content (string or bytes). Mutually exclusive with source_path.
+            source_path (Optional[str]): The local path of the file to attach. Mutually exclusive with content.
+            category (Optional[str]): Optional category for the attachment in the context of this job.
+
+        Returns:
+            Union[uuid.UUID, str]: The attachment key if uploaded, or the temp file path if written locally.
+
+        Raises:
+            ValueError: If neither content nor source_path is provided, or if both are provided.
+        """
+        if not (content is not None or source_path is not None):
+            raise ValueError("Either content or source_path must be provided.")
+        if content is not None and source_path is not None:
+            raise ValueError(
+                "Parameters content and source_path are mutually exclusive."
+            )
+
+        job_key_str = self._execution_context._instance_key
+        if job_key_str:
+            attachment_key = self._attachments_service.upload(
+                name=name, content=content, source_path=source_path
+            )
+            self._jobs_service.link_attachment(
+                attachment_key=attachment_key,
+                job_key=uuid.UUID(job_key_str),
+                category=category,
+            )
+            return attachment_key
+        else:
+            temp_dir = tempfile.gettempdir()
+            temp_path = os.path.join(temp_dir, name)
+            if source_path:
+                shutil.copyfile(source_path, temp_path)
+            elif content is not None:  # content can be str or bytes
+                mode = "w" if isinstance(content, str) else "wb"
+                with open(temp_path, mode) as f:
+                    f.write(content)
+            return temp_path
+
+    def download_attachment_from_current_job(
+        self,
+        *,
+        name: str,
+        destination_path: str,
+    ) -> str:
+        """Download an attachment from the current job or from temp storage.
+
+        If a job key is present in the execution context, it attempts to find an attachment with the given name by listing attachments for the current job (using JobsService) and then downloads it using AttachmentsService. If no job key, copies the file from the temp directory to the destination path.
+
+        Args:
+            name (str): The name of the attachment file.
+            destination_path (str): The local path where the attachment will be saved.
+
+        Returns:
+            str: The path to the downloaded file.
+
+        Raises:
+            FileNotFoundError: If the attachment is not found for the current job, or if the temp file does not exist.
+        """
+        job_key_str = self._execution_context._instance_key
+        if job_key_str:
+            job_attachments = self._jobs_service.list_attachments(
+                job_key=uuid.UUID(job_key_str)
+            )
+            match = next((a for a in job_attachments if a.name == name), None)
+            if not match:
+                raise FileNotFoundError(
+                    f"Attachment '{name}' not found for job {job_key_str}"
+                )
+            self._attachments_service.download(
+                key=match.key, destination_path=destination_path
+            )
+            return destination_path
+        else:
+            temp_dir = tempfile.gettempdir()
+            temp_path = os.path.join(temp_dir, name)
+            if not os.path.exists(temp_path):
+                raise FileNotFoundError(f"Temp file '{temp_path}' does not exist")
+            shutil.copyfile(temp_path, destination_path)
+            return destination_path

src/uipath/_uipath.py

@@ -15,6 +15,7 @@
     ConnectionsService,
     ContextGroundingService,
     FolderService,
+    JobAttachmentsService,
     JobsService,
     ProcessesService,
     QueuesService,
@@ -122,3 +123,8 @@ def folders(self) -> FolderService:
         if not self._folders_service:
             self._folders_service = FolderService(self._config, self._execution_context)
         return self._folders_service
+
+    @property
+    def job_attachments(self) -> JobAttachmentsService:
+        """Provides access to job attachment operations."""
+        return JobAttachmentsService(self._config, self._execution_context)