Executor / file_uploader.py
Soumik Bose
ok
2eac4ba
Raw
History Blame Contribute Delete
1.23 kB
import os
import requests
from typing import Optional
class ExternalFileUploader:
def __init__(self):
self.base_url = os.getenv("FILE_UPLOAD_API_URL")
self.api_key = os.getenv("FILE_UPLOAD_API_KEY")
def upload_file(self, file_bytes: bytes, file_name: str) -> Optional[str]:
try:
files = {
"files": (file_name, file_bytes, "image/png")
}
headers = {
"X-API-Key": self.api_key
}
response = requests.post(
f"{self.base_url}/api/v1/files/bulk-upload",
headers=headers,
files=files,
timeout=30
)
if response.status_code != 201:
print("Upload failed:", response.text)
return None
data = response.json()
# Expecting array response
if not isinstance(data, list) or not data:
return None
first = data[0]
if not first.get("success"):
return None
return first["data"]["file_url"]
except Exception as e:
print("Uploader error:", str(e))
return None