creates APIHandler methods for downloading attachments
method 'download_attachments_data" works with elabapi.UploadsApi() class to download binary data and other metadata of our files. CURRENTLY it downloads every single attachment which is not intended and it's only for testing purposes "download_attachments_to_disk" saves binary data to "output/attachments"
This commit is contained in:
@@ -1,41 +1,126 @@
|
||||
import requests
|
||||
import os, requests
|
||||
import elabapi_python as elabapi
|
||||
|
||||
|
||||
class APIHandler:
|
||||
'''
|
||||
"""
|
||||
Class to standardize the format of the headers of our http requests.
|
||||
'''
|
||||
"""
|
||||
|
||||
# TO-DO: remove static url.
|
||||
def __init__(self, apikey="", ELABFTW_API_URL="https://elabftw.fisica.unina.it/api/v2"):
|
||||
'''Init method, apikey suggested but not required (empty by default).'''
|
||||
self.auth = {"Authorization" : apikey}
|
||||
self.content = {"Content-Type" : "application/json"}
|
||||
def __init__(
|
||||
self, api_key="", ELABFTW_API_URL="https://elabftw.fisica.unina.it/api/v2"
|
||||
):
|
||||
"""Init method, apikey suggested but not required (empty by default)."""
|
||||
self.api_key = api_key
|
||||
self.auth = {"Authorization": api_key}
|
||||
self.content = {"Content-Type": "application/json"}
|
||||
self.header = {**self.auth, **self.content}
|
||||
self.elaburl = ELABFTW_API_URL
|
||||
|
||||
def get_entry_from_elabid(self, elabid, entryType="items"):
|
||||
'''
|
||||
"""
|
||||
Method which returns a resource's raw data (as dictionary) from its elabid and entry type.
|
||||
|
||||
Entry type can be either "experiments" or "items".
|
||||
'''
|
||||
"""
|
||||
# TO-DO: validation and error handling on entryType value.
|
||||
header = self.header
|
||||
response = requests.get(
|
||||
headers = header,
|
||||
url = f"{self.elaburl}/{entryType}/{elabid}",
|
||||
verify=True
|
||||
headers=header, url=f"{self.elaburl}/{entryType}/{elabid}", verify=True
|
||||
)
|
||||
if response.status_code // 100 in [1,2,3]:
|
||||
if response.status_code // 100 in [1, 2, 3]:
|
||||
entry_data = response.json()
|
||||
return entry_data
|
||||
elif response.status_code // 100 == 4:
|
||||
match response.status_code:
|
||||
case 401|403:
|
||||
raise ConnectionError(f"Invalid API key, authentication method or elabid. Check if an item with ID = {elabid} actually exists.")
|
||||
case 401 | 403:
|
||||
raise ConnectionError(
|
||||
f"Invalid API key, authentication method or elabid. Check if an item with ID = {elabid} actually exists."
|
||||
)
|
||||
case 404:
|
||||
raise ConnectionError(f"404: Not Found. This means there's no resource with this elabid (wrong elabid?) on your eLabFTW (wrong endpoint?).")
|
||||
raise ConnectionError(
|
||||
f"404: Not Found. This means there's no resource with this elabid (wrong elabid?) on your eLabFTW (wrong endpoint?)."
|
||||
)
|
||||
case 400:
|
||||
raise ConnectionError(f"400: Bad Request. This means the API endpoint you tried to reach is invalid. Did you tamper with the source code? If not, contact the developer.")
|
||||
raise ConnectionError(
|
||||
f"400: Bad Request. This means the API endpoint you tried to reach is invalid. Did you tamper with the source code? If not, contact the developer."
|
||||
)
|
||||
case _:
|
||||
raise ConnectionError(f"HTTP request failed with status code: {response.status_code} (NOTE: 4xx means user's fault).")
|
||||
raise ConnectionError(
|
||||
f"HTTP request failed with status code: {response.status_code} (NOTE: 4xx means user's fault)."
|
||||
)
|
||||
else:
|
||||
raise ConnectionError(f"There's a problem on the server. Status code: {response.status_code}.")
|
||||
raise ConnectionError(
|
||||
f"There's a problem on the server. Status code: {response.status_code}."
|
||||
)
|
||||
|
||||
def download_attachments_data(self, elabid, entryType="experiments"):
|
||||
"""
|
||||
Downloads attachments of a certain eLabFTW experiment (default) or item.
|
||||
Only returns their binary data. Use method download_attachments_to_disk to save to file.
|
||||
NOTE: Output is a dictionary where:
|
||||
* The keys are the attachments' filenames;
|
||||
* The values are the binary data for those attachments.
|
||||
|
||||
Args:
|
||||
elabid: eLabFTW internal ID of the selected resource.
|
||||
entryType: Resource type. Anything other than "experiments" or "items" WILL raise an error.
|
||||
"""
|
||||
if entryType not in ["experiments", "items"]:
|
||||
raise Exception(
|
||||
"You can only download attachments from experiments or items."
|
||||
)
|
||||
|
||||
config = elabapi.Configuration()
|
||||
config.api_key["api_key"] = api_key
|
||||
config.api_key_prefix["api_key"] = "Authorization"
|
||||
config.host = self.elaburl
|
||||
config.debug = False
|
||||
api_client = elabapi.ApiClient(config)
|
||||
api_client.set_default_header(
|
||||
header_name="Authorization", header_value=self.api_key
|
||||
)
|
||||
uploads_api = elabapi.UploadsApi(api_client)
|
||||
|
||||
# Actual uploads (dictionary):
|
||||
uploads = {
|
||||
upload.real_name: uploads_api.read_upload(
|
||||
entryType, elabid, upload.id, format="binary", _preload_content=False
|
||||
).data
|
||||
for upload in uploads_api.read_uploads(entryType, elabid)
|
||||
}
|
||||
|
||||
return uploads
|
||||
|
||||
def download_attachments_to_disk(
|
||||
self,
|
||||
elabid,
|
||||
entryType="experiments",
|
||||
dump_dir="output/attachments",
|
||||
persistent=True,
|
||||
):
|
||||
"""
|
||||
Downloads attachments of a certain eLabFTW experiment (default) or item.
|
||||
Downloads their binary data through method download_attachments_data and dumps it to dump_dir.
|
||||
|
||||
Args:
|
||||
elabid: eLabFTW internal ID of the selected resource.
|
||||
entryType: Resource type. Anything other than "experiments" or "items" WILL raise an error.
|
||||
dump_dir: Directory to which to save the attachments. Default is "output/attachments".
|
||||
persistent: [Unused] Decides if the files will stay on disk after all operations are completed.
|
||||
If set to False, deletes the file upon exiting.
|
||||
"""
|
||||
|
||||
if entryType not in ["experiments", "items"]:
|
||||
raise Exception(
|
||||
"You can only download attachments from experiments or items."
|
||||
)
|
||||
|
||||
uploads = download_attachments_data(elabid, entryType=entryType)
|
||||
for file in uploads:
|
||||
raw_data = uploads["file"]
|
||||
with open(os.path.join(dump_dir, f"exp{elabid}-{file}"), "wb") as f:
|
||||
f.write(raw_data)
|
||||
return
|
||||
|
||||
|
||||
Reference in New Issue
Block a user