1import json
2import requests
3from typing import Optional
5def create_signed_upload_url(url: str, size: int, user_id: str, memobin_api_key: str) -> str:
6 """Create a signed upload URL for memobin.
8 Args:
9 url: The target URL for the file
10 size: Size of the file in bytes
11 user_id: User ID for memobin
12 memobin_api_key: API key for memobin authentication
14 Returns:
15 The signed upload URL
17 Raises:
18 ValueError: If the URL prefix is invalid
19 requests.RequestException: If the API request fails
20 """
21 prefix = "https://tempory.net/f/memobin/"
22 if not url.startswith(prefix):
23 raise ValueError("Invalid url. Does not have proper prefix")
25 file_path = url[len(prefix):]
26 tempory_api_url = "https://hub.tempory.net/api/uploadFile"
28 response = requests.post(
29 tempory_api_url,
30 headers={
31 "Content-Type": "application/json",
32 "Authorization": f"Bearer {memobin_api_key}"
33 },
34 json={
35 "appName": "memobin",
36 "filePath": file_path,
37 "size": size,
38 "userId": user_id
39 }
40 )
42 if not response.ok:
43 raise requests.RequestException("Failed to get signed url")
45 result = response.json()
46 upload_url = result["uploadUrl"]
47 download_url = result["downloadUrl"]
49 if download_url != url:
50 raise ValueError("Mismatch between download url and url")
52 return upload_url
54def construct_memobin_url(alg_name: str, dataset_name: str, alg_version: str,
55 dataset_version: str, system_version: str) -> str:
56 """Construct the memobin URL for a specific benchmark result.
58 Args:
59 alg_name: Name of the algorithm
60 dataset_name: Name of the dataset
61 alg_version: Version of the algorithm
62 dataset_version: Version of the dataset
63 system_version: Version of the system
65 Returns:
66 The constructed memobin URL
67 """
68 path = f"{alg_name}/{dataset_name}/{alg_version}/{dataset_version}/{system_version}/metadata.json"
69 return f"https://tempory.net/f/memobin/{path}"
71def upload_to_memobin(metadata: dict, url: str, user_id: str, memobin_api_key: str) -> None:
72 """Upload metadata to memobin.
74 Args:
75 metadata: The metadata to upload
76 url: The target URL for the file
77 user_id: User ID for memobin
78 memobin_api_key: API key for memobin authentication
80 Raises:
81 requests.RequestException: If the upload fails
82 """
83 metadata_bytes = json.dumps(metadata).encode('utf-8')
84 size = len(metadata_bytes)
86 upload_url = create_signed_upload_url(url, size, user_id, memobin_api_key)
88 response = requests.put(
89 upload_url,
90 data=metadata_bytes,
91 headers={"Content-Type": "application/json"}
92 )
94 if not response.ok:
95 raise requests.RequestException("Failed to upload metadata to memobin")
97def download_from_memobin(url: str) -> Optional[dict]:
98 """Download metadata from memobin.
100 Args:
101 url: The URL to download from
103 Returns:
104 The downloaded metadata as a dictionary, or None if not found
106 Raises:
107 requests.RequestException: If the download fails for a reason other than 404
108 """
109 response = None
110 try:
111 response = requests.get(url)
112 if response.status_code == 404:
113 return None
114 response.raise_for_status()
115 return response.json()
116 except requests.RequestException as e:
117 if response and response.status_code == 404:
118 return None
119 raise e