Den 429 Add upload and download to dendrite (#73)

Author: coffecup25

dendrite/async_api/_core/dendrite_browser.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
+import pathlib
 import re
-from typing import Any, List, Literal, Optional, Union
+from typing import Any, List, Literal, Optional, Sequence, Union
 from uuid import uuid4
 import os
 from loguru import logger
@@ -11,6 +12,7 @@
     FileChooser,
     Download,
     Error,
+    FilePayload,
 )
 
 from dendrite.async_api._api.dto.authenticate_dto import AuthenticateDTO
@@ -374,7 +376,7 @@ async def _get_active_page_manager(self) -> PageManager:
 
         return self._active_page_manager
 
-    async def _get_download(self, pw_page: PlaywrightPage, timeout: float) -> Download:
+    async def get_download(self, pw_page: PlaywrightPage, timeout: float) -> Download:
         """
         Retrieves the download event from the browser.
 
@@ -386,6 +388,33 @@ async def _get_download(self, pw_page: PlaywrightPage, timeout: float) -> Downlo
         """
         return await self._impl.get_download(self, pw_page, timeout)
 
+    async def upload_files(
+        self,
+        files: Union[
+            str,
+            pathlib.Path,
+            FilePayload,
+            Sequence[Union[str, pathlib.Path]],
+            Sequence[FilePayload],
+        ],
+        timeout: float = 30000,
+    ) -> None:
+        """
+        Uploads files to the active page using a file chooser.
+
+        Args:
+            files (Union[str, pathlib.Path, FilePayload, Sequence[Union[str, pathlib.Path]], Sequence[FilePayload]]): The file(s) to be uploaded.
+                This can be a file path, a `FilePayload` object, or a sequence of file paths or `FilePayload` objects.
+            timeout (float, optional): The maximum amount of time (in milliseconds) to wait for the file chooser to be ready. Defaults to 30.
+
+        Returns:
+            None
+        """
+        page = await self.get_active_page()
+
+        file_chooser = await self._get_filechooser(page.playwright_page, timeout)
+        await file_chooser.set_files(files)
+
     async def _get_filechooser(
         self, pw_page: PlaywrightPage, timeout: float = 30000
     ) -> FileChooser:

dendrite/async_api/_core/dendrite_page.py

@@ -153,7 +153,7 @@ async def get_download(self, timeout: float = 30000) -> Download:
         Returns:
             The downloaded file data.
         """
-        return await self.dendrite_browser._get_download(self.playwright_page, timeout)
+        return await self.dendrite_browser.get_download(self.playwright_page, timeout)
 
     def _get_context(self, element: Any) -> Union[PlaywrightPage, FrameLocator]:
         """

dendrite/sync_api/_core/dendrite_browser.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
+import pathlib
 import re
-from typing import Any, List, Literal, Optional, Union
+from typing import Any, List, Literal, Optional, Sequence, Union
 from uuid import uuid4
 import os
 from loguru import logger
@@ -11,6 +12,7 @@
     FileChooser,
     Download,
     Error,
+    FilePayload,
 )
 from dendrite.sync_api._api.dto.authenticate_dto import AuthenticateDTO
 from dendrite.sync_api._api.dto.upload_auth_session_dto import UploadAuthSessionDTO
@@ -333,11 +335,11 @@ def _get_active_page_manager(self) -> PageManager:
             Exception: If there is an issue launching the browser or retrieving the PageManager.
         """
         if not self._active_page_manager:
-            _, _, active_page_manager = self._launch()
+            (_, _, active_page_manager) = self._launch()
             return active_page_manager
         return self._active_page_manager
 
-    def _get_download(self, pw_page: PlaywrightPage, timeout: float) -> Download:
+    def get_download(self, pw_page: PlaywrightPage, timeout: float) -> Download:
         """
         Retrieves the download event from the browser.
 
@@ -349,6 +351,32 @@ def _get_download(self, pw_page: PlaywrightPage, timeout: float) -> Download:
         """
         return self._impl.get_download(self, pw_page, timeout)
 
+    def upload_files(
+        self,
+        files: Union[
+            str,
+            pathlib.Path,
+            FilePayload,
+            Sequence[Union[str, pathlib.Path]],
+            Sequence[FilePayload],
+        ],
+        timeout: float = 30000,
+    ) -> None:
+        """
+        Uploads files to the active page using a file chooser.
+
+        Args:
+            files (Union[str, pathlib.Path, FilePayload, Sequence[Union[str, pathlib.Path]], Sequence[FilePayload]]): The file(s) to be uploaded.
+                This can be a file path, a `FilePayload` object, or a sequence of file paths or `FilePayload` objects.
+            timeout (float, optional): The maximum amount of time (in milliseconds) to wait for the file chooser to be ready. Defaults to 30.
+
+        Returns:
+            None
+        """
+        page = self.get_active_page()
+        file_chooser = self._get_filechooser(page.playwright_page, timeout)
+        file_chooser.set_files(files)
+
     def _get_filechooser(
         self, pw_page: PlaywrightPage, timeout: float = 30000
     ) -> FileChooser:

dendrite/sync_api/_core/dendrite_page.py

@@ -123,7 +123,7 @@ def get_download(self, timeout: float = 30000) -> Download:
         Returns:
             The downloaded file data.
         """
-        return self.dendrite_browser._get_download(self.playwright_page, timeout)
+        return self.dendrite_browser.get_download(self.playwright_page, timeout)
 
     def _get_context(self, element: Any) -> Union[PlaywrightPage, FrameLocator]:
         """

dendrite/sync_api/_dom/util/mild_strip.py

@@ -26,7 +26,7 @@ def _mild_strip(soup: BeautifulSoup, keep_d_id: bool = True) -> None:
             continue
         tag.attrs = {
             attr: value[:100] if isinstance(value, str) else value
-            for attr, value in tag.attrs.items()
+            for (attr, value) in tag.attrs.items()
         }
         if keep_d_id == False:
             del tag["d-id"]