1import json
2import requests
3from typing import Optional
6def create_signed_upload_url(
7 url: str, size: int, user_id: str, memobin_api_key: str
8) -> str:
9 """Create a signed upload URL for memobin.
11 Args:
12 url: The target URL for the file
13 size: Size of the file in bytes
14 user_id: User ID for memobin
15 memobin_api_key: API key for memobin authentication
17 Returns:
18 The signed upload URL
20 Raises:
21 ValueError: If the URL prefix is invalid
22 requests.RequestException: If the API request fails
23 """
24 prefix = "https://tempory.net/f/memobin/"
25 if not url.startswith(prefix):
26 raise ValueError("Invalid url. Does not have proper prefix")
28 file_path = url[len(prefix) :]
29 tempory_api_url = "https://hub.tempory.net/api/uploadFile"
31 response = requests.post(
32 tempory_api_url,
33 headers={
34 "Content-Type": "application/json",
35 "Authorization": f"Bearer {memobin_api_key}",
36 },
37 json={
38 "appName": "memobin",
39 "filePath": file_path,
40 "size": size,
41 "userId": user_id,
42 },
43 )
45 if not response.ok:
46 raise requests.RequestException("Failed to get signed url")
48 result = response.json()
49 upload_url = result["uploadUrl"]
50 download_url = result["downloadUrl"]
52 if download_url != url:
53 raise ValueError("Mismatch between download url and url")
55 return upload_url
58def construct_memobin_url(
59 alg_name: str,
60 dataset_name: str,
61 alg_version: str,
62 dataset_version: str,
63 system_version: str,
64 file_type: str = "metadata.json",
65) -> str:
66 """Construct the memobin URL for a specific benchmark result or dataset.
68 Args:
69 alg_name: Name of the algorithm
70 dataset_name: Name of the dataset
71 alg_version: Version of the algorithm
72 dataset_version: Version of the dataset
73 system_version: Version of the system
74 file_type: Type of file (metadata.json or data.bin)
76 Returns:
77 The constructed memobin URL
78 """
79 path = f"{alg_name}/{dataset_name}/{alg_version}/{dataset_version}/{system_version}/{file_type}"
80 return f"https://tempory.net/f/memobin/{path}"
83def construct_dataset_url(
84 dataset_name: str, dataset_version: str, format: str = "dat"
85) -> str:
86 """Construct the memobin URL for a dataset.
88 Args:
89 dataset_name: Name of the dataset
90 dataset_version: Version of the dataset
91 format: File format ("dat", "npy", or "json")
93 Returns:
94 The constructed memobin URL for the dataset
95 """
96 path = f"datasets/{dataset_name}/{dataset_version}/{dataset_name}-{dataset_version}.{format}"
97 return f"https://tempory.net/f/memobin/{path}"
100def upload_to_memobin(
101 data: dict | bytes,
102 url: str,
103 memobin_api_key: str,
104 content_type: str = "application/json",
105) -> None:
106 """Upload data to memobin.
108 Args:
109 data: The data to upload (dict for JSON or bytes for binary)
110 url: The target URL for the file
111 memobin_api_key: API key for memobin authentication
112 content_type: Content type of the data
114 Raises:
115 requests.RequestException: If the upload fails
116 """
117 if isinstance(data, dict):
118 data_bytes = json.dumps(data).encode("utf-8")
119 else:
120 data_bytes = data
121 size = len(data_bytes)
123 upload_url = create_signed_upload_url(url, size, "zia", memobin_api_key)
125 response = requests.put(
126 upload_url, data=data_bytes, headers={"Content-Type": content_type}
127 )
129 if not response.ok:
130 raise requests.RequestException("Failed to upload data to memobin")
133def exists_in_memobin(url: str) -> bool:
134 """Check if a file exists in memobin using a HEAD request.
136 Args:
137 url: The URL to check
139 Returns:
140 True if the file exists, False otherwise
141 """
142 try:
143 response = requests.head(url)
144 return (
145 200 <= response.status_code < 300
146 ) # Any 2xx status code indicates success
147 except requests.RequestException:
148 return False
151def download_from_memobin(url: str, as_json: bool = True) -> Optional[dict | bytes]:
152 """Download data from memobin.
154 Args:
155 url: The URL to download from
156 as_json: Whether to parse the response as JSON
158 Returns:
159 The downloaded data as a dictionary or bytes, or None if not found
161 Raises:
162 requests.RequestException: If the download fails for a reason other than 404
163 """
164 response = None
165 try:
166 response = requests.get(url)
167 if response.status_code == 404:
168 return None
169 response.raise_for_status()
170 return response.json() if as_json else response.content
171 except requests.RequestException as e:
172 if response and response.status_code == 404:
173 return None
174 raise e