change storage to non-async for now

Author: cjber

python/thirdweb-ai/src/thirdweb_ai/services/storage.py

@@ -1,28 +1,25 @@
-import asyncio
 import hashlib
 import json
 import mimetypes
 import os
-from collections.abc import AsyncGenerator
 from dataclasses import asdict, is_dataclass
 from io import BytesIO
 from pathlib import Path
 from typing import Annotated, Any
 
-import httpx
+import requests
 from pydantic import BaseModel
 
 from thirdweb_ai.services.service import Service
 from thirdweb_ai.tools.tool import tool
 
 
-async def async_read_file_chunks(file_path: str | Path, chunk_size: int = 8192) -> AsyncGenerator[bytes, None]:
-    """Read file in chunks asynchronously to avoid loading entire file into memory."""
-    async with asyncio.Lock():
-        path_obj = Path(file_path) if isinstance(file_path, str) else file_path
-        with path_obj.open("rb") as f:
-            while chunk := f.read(chunk_size):
-                yield chunk
+def read_file_chunks(file_path: str | Path, chunk_size: int = 8192):
+    """Read file in chunks to avoid loading entire file into memory."""
+    path_obj = Path(file_path) if isinstance(file_path, str) else file_path
+    with path_obj.open("rb") as f:
+        while chunk := f.read(chunk_size):
+            yield chunk
 
 
 class Storage(Service):
@@ -46,18 +43,44 @@ def fetch_ipfs_content(
 
         ipfs_hash = ipfs_hash.removeprefix("ipfs://")
         path = f"https://{self.gateway_url}.{self.gateway_hostname}/ipfs/{ipfs_hash}"
-        return self._get(path)
+        return self._get_file(path)
+
+    def _get_file(self, path: str, params: dict[str, Any] | None = None, headers: dict[str, Any] | None = None):
+        _headers = self._make_headers()
+        if headers:
+            _headers.update(headers)
+        response = self.client.get(path, params=params, headers=_headers)
+        response.raise_for_status()
+        
+        content_type = response.headers.get("Content-Type", "")
+        
+        # Handle JSON responses
+        if "application/json" in content_type:
+            return response.json()
+        
+        # Handle binary files (images, pdfs, etc)
+        if content_type.startswith(("image/", "application/pdf", "application/octet-stream")):
+            return {"content": response.content, "content_type": content_type}
+            
+        # Handle text content (html, plain text, etc)
+        if content_type.startswith(("text/", "application/xml")):
+            return {"content": response.text, "content_type": content_type}
+            
+        # Default fallback - try json first, then return content with type
+        try:
+            return response.json()
+        except ValueError:
+            return {"content": response.content, "content_type": content_type}
 
-    async def _async_post_file(self, url: str, files: dict[str, Any]) -> dict[str, Any]:
-        """Post files to a URL using async client with proper authorization headers."""
+    def _post_file(self, url: str, files: dict[str, Any]) -> dict[str, Any]:
+        """Post files to a URL with proper authorization headers."""
         headers = self._make_headers()
-        # Remove the Content-Type as httpx will set it correctly for multipart/form-data
+        # Remove the Content-Type as requests will set it correctly for multipart/form-data
         headers.pop("Content-Type", None)
 
-        async with httpx.AsyncClient() as client:
-            response = await client.post(url, files=files, headers=headers)
-            response.raise_for_status()
-            return response.json()
+        response = requests.post(url, files=files, headers=headers)
+        response.raise_for_status()
+        return response.json()
 
     def _is_json_serializable(self, data: Any) -> bool:
         """Check if data is JSON serializable (dict, dataclass, or BaseModel)."""
@@ -80,7 +103,7 @@ def _is_valid_path(self, path: str) -> bool:
         """Check if the string is a valid file or directory path."""
         return Path(path).exists()
 
-    async def _prepare_directory_files(
+    def _prepare_directory_files(
         self, directory_path: Path, chunk_size: int = 8192
     ) -> list[tuple[str, BytesIO, str]]:
         """
@@ -98,7 +121,7 @@ async def _prepare_directory_files(
 
                 # Create a buffer and read the file in chunks
                 buffer = BytesIO()
-                async for chunk in async_read_file_chunks(file_path, chunk_size):
+                for chunk in read_file_chunks(file_path, chunk_size):
                     buffer.write(chunk)
                 buffer.seek(0)  # Reset buffer position
 
@@ -109,7 +132,7 @@ async def _prepare_directory_files(
     @tool(
         description="Upload a file, directory, or JSON data to IPFS. Stores any type on decentralized storage and returns an IPFS URI."
     )
-    async def upload_to_ipfs(
+    def upload_to_ipfs(
         self,
         data: Annotated[
             Any, "Data to upload: can be a file path, directory path, dict, dataclass, or BaseModel instance."
@@ -133,7 +156,7 @@ async def upload_to_ipfs(
         if self._is_json_serializable(data):
             json_content = self._convert_to_json(data)
             files = {"file": ("data.json", BytesIO(json_content.encode()), "application/json")}
-            body = await self._async_post_file(storage_url, files)
+            body = self._post_file(storage_url, files)
             return f"ipfs://{body['IpfsHash']}"
 
         # Handle string paths to files or directories
@@ -146,18 +169,18 @@ async def upload_to_ipfs(
 
                 # Create a buffer to hold chunks for streaming upload
                 buffer = BytesIO()
-                async for chunk in async_read_file_chunks(path):
+                for chunk in read_file_chunks(path):
                     buffer.write(chunk)
 
                 buffer.seek(0)  # Reset buffer position
                 files = {"file": (path.name, buffer, content_type)}
-                body = await self._async_post_file(storage_url, files)
+                body = self._post_file(storage_url, files)
                 return f"ipfs://{body['IpfsHash']}"
 
             # Directory upload - preserve directory structure
             if path.is_dir():
                 # Prepare all files from the directory with preserved structure
-                files_data = await self._prepare_directory_files(path)
+                files_data = self._prepare_directory_files(path)
 
                 if not files_data:
                     raise ValueError(f"Directory is empty: {data}")
@@ -166,15 +189,15 @@ async def upload_to_ipfs(
                     f"file{i}": (relative_path, buffer, content_type)
                     for i, (relative_path, buffer, content_type) in enumerate(files_data)
                 }
-                body = await self._async_post_file(storage_url, files_dict)
+                body = self._post_file(storage_url, files_dict)
                 return f"ipfs://{body['IpfsHash']}"
 
             raise ValueError(f"Path exists but is neither a file nor a directory: {data}")
 
         try:
             content_type = mimetypes.guess_type(data)[0] or "application/octet-stream"
             files = {"file": ("data.txt", BytesIO(data.encode()), content_type)}
-            body = await self._async_post_file(storage_url, files)
+            body = self._post_file(storage_url, files)
             return f"ipfs://{body['IpfsHash']}"
         except TypeError as e:
             raise TypeError(