make library asynchronous

pyzxz/pyzxz.py

@@ -1,11 +1,11 @@
-import requests
+import aiohttp
+import asyncio
 from pathlib import Path
 from typing import Union
 
-
 class ZeroXZero:
     """
-    A static utility class for interacting with the 0x0.st file hosting service.
+    An async utility class for interacting with the 0x0.st file hosting service.
 
     Features:
     - Upload files from disk
@@ -20,48 +20,51 @@ class ZeroXZero:
     }
 
     @staticmethod
-    def upload(file_path: Union[str, Path]) -> str:
+    async def upload(file_path: Union[str, Path]) -> str:
         file_path = Path(file_path)
 
         if not file_path.exists():
             raise FileNotFoundError(f"No such file: {file_path}")
 
-        with file_path.open("rb") as f:
-            response = requests.post(
+        async with aiohttp.ClientSession(headers=ZeroXZero.HEADERS) as session:
+            with file_path.open("rb") as f:
+                data = aiohttp.FormData()
+                data.add_field("file", f, filename=file_path.name)
+                async with session.post(
+                    ZeroXZero.ENDPOINT_URL,
+                    data=data
+                ) as response:
+                    text = await response.text()
+                    if response.status == 200 and text.startswith("https://"):
+                        return text.strip()
+                    raise ValueError(f"Upload failed: {text.strip()}")
+
+    @staticmethod
+    async def upload_from_bytes(data: bytes, filename: str) -> str:
+        async with aiohttp.ClientSession(headers=ZeroXZero.HEADERS) as session:
+            form = aiohttp.FormData()
+            form.add_field("file", data, filename=filename)
+            async with session.post(
                 ZeroXZero.ENDPOINT_URL,
-                files={"file": f},
-                headers=ZeroXZero.HEADERS
-            )
-
-        if response.ok and response.text.startswith("https://"):
-            return response.text.strip()
-        raise ValueError(f"Upload failed: {response.text.strip()}")
+                data=form
+            ) as response:
+                text = await response.text()
+                if response.status == 200 and text.startswith("https://"):
+                    return text.strip()
+                raise ValueError(f"Upload failed: {text.strip()}")
 
     @staticmethod
-    def upload_from_bytes(data: bytes, filename: str) -> str:
-        files = {"file": (filename, data)}
-        response = requests.post(
-            ZeroXZero.ENDPOINT_URL,
-            files=files,
-            headers=ZeroXZero.HEADERS
-        )
-
-        if response.ok and response.text.startswith("https://"):
-            return response.text.strip()
-        raise ValueError(f"Upload failed: {response.text.strip()}")
+    async def upload_text(text: str, filename: str = "text.txt") -> str:
+        return await ZeroXZero.upload_from_bytes(text.encode("utf-8"), filename)
 
     @staticmethod
-    def upload_text(text: str, filename: str = "text.txt") -> str:
-        return ZeroXZero.upload_from_bytes(text.encode("utf-8"), filename)
-
-    @staticmethod
-    def is_available() -> bool:
+    async def is_available() -> bool:
         try:
-            response = requests.head(
-                ZeroXZero.ENDPOINT_URL,
-                timeout=3,
-                headers=ZeroXZero.HEADERS
-            )
-            return response.status_code == 200
-        except requests.RequestException:
-            return False
+            async with aiohttp.ClientSession(headers=ZeroXZero.HEADERS) as session:
+                async with session.head(
+                    ZeroXZero.ENDPOINT_URL,
+                    timeout=aiohttp.ClientTimeout(total=3)
+                ) as response:
+                    return response.status == 200
+        except Exception:
+            return False