feat: integrate aiofiles for asynchronous file handling and update dependencies

Author: AdirAmsalem

decart_sdk/client.py

@@ -1,4 +1,5 @@
 from typing import Any, Optional
+import aiohttp
 from pydantic import BaseModel, Field, field_validator, ValidationError
 from .errors import InvalidAPIKeyError, InvalidBaseURLError, InvalidInputError
 from .models import ModelDefinition
@@ -39,6 +40,27 @@ def __init__(self, api_key: str, base_url: str = "https://api.decart.ai") -> Non
 
         self.api_key = api_key
         self.base_url = base_url
+        self._session: Optional[aiohttp.ClientSession] = None
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        """Get or create the aiohttp session."""
+        if self._session is None or self._session.closed:
+            timeout = aiohttp.ClientTimeout(total=300)
+            self._session = aiohttp.ClientSession(timeout=timeout)
+        return self._session
+
+    async def close(self) -> None:
+        """Close the HTTP session and cleanup resources."""
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    async def __aenter__(self):
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit."""
+        await self.close()
 
     async def process(self, options: dict[str, Any]) -> bytes:
         """
@@ -62,29 +84,33 @@ async def process(self, options: dict[str, Any]) -> bytes:
 
         inputs = {k: v for k, v in options.items() if k not in ("model", "cancel_token")}
 
-        # Separate file inputs from other inputs
-        file_fields = {"data", "start", "end"}
-        file_inputs = {k: v for k, v in inputs.items() if k in file_fields}
-        non_file_inputs = {k: v for k, v in inputs.items() if k not in file_fields}
+        # File fields that need special handling (not validated by Pydantic)
+        FILE_FIELDS = {"data", "start", "end"}
+        
+        # Separate file inputs from regular inputs
+        file_inputs = {k: v for k, v in inputs.items() if k in FILE_FIELDS}
+        non_file_inputs = {k: v for k, v in inputs.items() if k not in FILE_FIELDS}
 
-        # Validate inputs using model's schema
-        validation_inputs = non_file_inputs.copy()
-        for field in file_fields:
-            if field in file_inputs:
-                validation_inputs[field] = b""  # Placeholder for validation
+        # Validate non-file inputs and create placeholder for file fields
+        validation_inputs = {
+            **non_file_inputs,
+            **{k: b"" for k in file_inputs.keys()}  # Placeholder bytes for validation
+        }
 
         try:
             validated_inputs = model.input_schema(**validation_inputs)
         except ValidationError as e:
             raise InvalidInputError(f"Invalid inputs for {model.name}: {str(e)}") from e
 
-        # Merge validated inputs with file inputs
-        processed_inputs = validated_inputs.model_dump(exclude_none=True)
-        for field in file_fields:
-            if field in file_inputs:
-                processed_inputs[field] = file_inputs[field]
+        # Build final inputs: validated non-file inputs + original file inputs
+        processed_inputs = {
+            **validated_inputs.model_dump(exclude_none=True),
+            **file_inputs  # Override placeholders with actual file data
+        }
 
+        session = await self._get_session()
         response = await send_request(
+            session=session,
             base_url=self.base_url,
             api_key=self.api_key,
             model=model,

decart_sdk/process/request.py

@@ -1,39 +1,83 @@
 import aiohttp
+import aiofiles
 import asyncio
+from pathlib import Path
 from typing import Any, Optional
 from ..types import FileInput
 from ..models import ModelDefinition
 from ..errors import InvalidInputError, ProcessingError
 
 
-async def file_input_to_bytes(input_data: FileInput) -> tuple[bytes, str]:
+async def file_input_to_bytes(
+    input_data: FileInput, session: aiohttp.ClientSession
+) -> tuple[bytes, str]:
+    """Convert various file input types to bytes asynchronously.
+    
+    Args:
+        input_data: The file input (bytes, Path, str, or file-like object)
+        session: Reusable aiohttp session for URL fetching
+        
+    Returns:
+        Tuple of (content bytes, content type)
+        
+    Raises:
+        InvalidInputError: If input is invalid or processing fails
+    """
+    
     if isinstance(input_data, bytes):
         return input_data, "application/octet-stream"
 
-    if hasattr(input_data, "read"):
+    if isinstance(input_data, Path):
+        # Async file reading with aiofiles
+        try:
+            async with aiofiles.open(input_data, mode="rb") as f:
+                content = await f.read()
+            return content, "application/octet-stream"
+        except FileNotFoundError:
+            raise InvalidInputError(f"File not found: {input_data}")
+        except Exception as e:
+            raise InvalidInputError(f"Failed to read file {input_data}: {str(e)}")
+
+    if isinstance(input_data, str):
+        # Check if it's a file path
+        path = Path(input_data)
+        if path.exists():
+            try:
+                async with aiofiles.open(path, mode="rb") as f:
+                    content = await f.read()
+                return content, "application/octet-stream"
+            except Exception as e:
+                raise InvalidInputError(f"Failed to read file {input_data}: {str(e)}")
+        
+        # Otherwise treat as URL
+        if not input_data.startswith(("http://", "https://")):
+            raise InvalidInputError(
+                f"Input must be a URL (http:// or https://) or existing file path: {input_data}"
+            )
+
+        # Use the provided session instead of creating a new one
+        async with session.get(input_data) as response:
+            if not response.ok:
+                raise InvalidInputError(
+                    f"Failed to fetch file from URL: {response.status}"
+                )
+            content = await response.read()
+            content_type = response.headers.get("Content-Type", "application/octet-stream")
+            return content, content_type
+
+    from ..types import HasRead
+    if isinstance(input_data, HasRead):
+        # Sync file-like objects (for backwards compatibility)
         content = await asyncio.to_thread(input_data.read)
         if isinstance(content, str):
             content = content.encode()
         return content, "application/octet-stream"
 
-    if isinstance(input_data, str):
-        if not input_data.startswith(("http://", "https://")):
-            raise InvalidInputError("URL must start with http:// or https://")
-
-        async with aiohttp.ClientSession() as session:
-            async with session.get(input_data) as response:
-                if not response.ok:
-                    raise InvalidInputError(
-                        f"Failed to fetch file from URL: {response.status}"
-                    )
-                content = await response.read()
-                content_type = response.headers.get("Content-Type", "application/octet-stream")
-                return content, content_type
-
-    raise InvalidInputError("Invalid file input type")
+    raise InvalidInputError(f"Invalid file input type: {type(input_data)}")
 
 
 async def send_request(
+    session: aiohttp.ClientSession,
     base_url: str,
     api_key: str,
     model: ModelDefinition,
@@ -45,28 +89,25 @@ async def send_request(
     for key, value in inputs.items():
         if value is not None:
             if key in ("data", "start", "end"):
-                content, content_type = await file_input_to_bytes(value)
+                content, content_type = await file_input_to_bytes(value, session)
                 form_data.add_field(key, content, content_type=content_type)
             else:
                 form_data.add_field(key, str(value))
 
     endpoint = f"{base_url}{model.url_path}"
 
-    timeout = aiohttp.ClientTimeout(total=300)
-
     async def make_request() -> bytes:
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.post(
-                endpoint,
-                headers={"X-API-KEY": api_key},
-                data=form_data,
-            ) as response:
-                if not response.ok:
-                    error_text = await response.text()
-                    raise ProcessingError(
-                        f"Processing failed: {response.status} - {error_text}"
-                    )
-                return await response.read()
+        async with session.post(
+            endpoint,
+            headers={"X-API-KEY": api_key},
+            data=form_data,
+        ) as response:
+            if not response.ok:
+                error_text = await response.text()
+                raise ProcessingError(
+                    f"Processing failed: {response.status} - {error_text}"
+                )
+            return await response.read()
 
     if cancel_token:
         request_task = asyncio.create_task(make_request())

decart_sdk/types.py

@@ -1,8 +1,16 @@
-from typing import BinaryIO, Union, Optional
+from typing import BinaryIO, Union, Optional, Protocol, runtime_checkable
+from pathlib import Path
 from pydantic import BaseModel, Field
 
 
-FileInput = Union[BinaryIO, bytes, str]
+@runtime_checkable
+class HasRead(Protocol):
+    """Protocol for file-like objects with a read method."""
+    def read(self) -> Union[bytes, str]:
+        ...
+
+
+FileInput = Union[HasRead, bytes, str, Path]
 
 
 class Prompt(BaseModel):