fix onedrive connector (#362)

Author: vangheem

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.4.4
+
+### Fixes
+
+* **Fix AsyncIO support for OneDrive connector**
+
 ## 0.4.3
 
 ### Enhancements

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "0.4.3"  # pragma: no cover
+__version__ = "0.4.4"  # pragma: no cover

unstructured_ingest/v2/interfaces/indexer.py

@@ -1,4 +1,4 @@
-from abc import ABC, abstractmethod
+from abc import ABC
 from typing import Any, AsyncGenerator, Generator, Optional, TypeVar
 
 from pydantic import BaseModel
@@ -22,9 +22,8 @@ class Indexer(BaseProcess, BaseConnector, ABC):
     def is_async(self) -> bool:
         return False
 
-    @abstractmethod
     def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
-        pass
+        raise NotImplementedError()
 
     async def run_async(self, **kwargs: Any) -> AsyncGenerator[FileData, None]:
         raise NotImplementedError()

unstructured_ingest/v2/processes/connectors/onedrive.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 from time import time
-from typing import TYPE_CHECKING, Any, AsyncIterator, Generator, Iterator, Optional, TypeVar
+from typing import TYPE_CHECKING, Any, AsyncIterator, Optional
 
 from dateutil import parser
 from pydantic import Field, Secret
@@ -101,27 +101,6 @@ class OnedriveIndexerConfig(IndexerConfig):
     recursive: bool = False
 
 
-T = TypeVar("T")
-
-
-def async_iterable_to_sync_iterable(iterator: AsyncIterator[T]) -> Iterator[T]:
-    # This version works on Python 3.9 by manually handling the async iteration.
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    try:
-        while True:
-            try:
-                # Instead of anext(iterator), we directly call __anext__().
-                # __anext__ returns a coroutine that we must run until complete.
-                future = iterator.__anext__()
-                result = loop.run_until_complete(future)
-                yield result
-            except StopAsyncIteration:
-                break
-    finally:
-        loop.close()
-
-
 @dataclass
 class OnedriveIndexer(Indexer):
     connection_config: OnedriveConnectionConfig
@@ -215,7 +194,10 @@ async def drive_item_to_file_data(self, drive_item: "DriveItem") -> FileData:
         # Offload the file data creation if it's not guaranteed async
         return await asyncio.to_thread(self.drive_item_to_file_data_sync, drive_item)
 
-    async def _run_async(self, **kwargs: Any) -> AsyncIterator[FileData]:
+    def is_async(self) -> bool:
+        return True
+
+    async def run_async(self, **kwargs: Any) -> AsyncIterator[FileData]:
         token_resp = await asyncio.to_thread(self.connection_config.get_token)
         if "error" in token_resp:
             raise SourceConnectionError(
@@ -230,12 +212,6 @@ async def _run_async(self, **kwargs: Any) -> AsyncIterator[FileData]:
             file_data = await self.drive_item_to_file_data(drive_item=drive_item)
             yield file_data
 
-    def run(self, **kwargs: Any) -> Generator[FileData, None, None]:
-        # Convert the async generator to a sync generator without loading all data into memory
-        async_gen = self._run_async(**kwargs)
-        for item in async_iterable_to_sync_iterable(async_gen):
-            yield item
-
 
 class OnedriveDownloaderConfig(DownloaderConfig):
     pass