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"

requirements.txt

@@ -2,3 +2,4 @@ requests
 asyncio
 h5py
 pillow
+elabapi_python

src/APIHandler.py

@@ -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
+

src/classes.py

@@ -1,8 +1,9 @@
 import os, json, requests
 from APIHandler import APIHandler
 
+
 class Layer:
-    '''
+    """
     Layer(layer_data) - where layer_data is a Python dictionary.
 
     Meant to be used for eLabFTW Experiments of the "PLD Deposition" category.
@@ -10,22 +11,27 @@ class Layer:
     eLabFTW experiments contain most of the data required by the NeXus file - although every layer is on a different eLab entry;
     unfortunately, some data like the target's chemical formula must be retrieved through additional HTTP requests.
     Attributes 'target_elabid', 'rheed_system_elabid' and 'laser_system_elabid' contain elabid's for these resources, which are all items.
-    '''
+    """
+
     def __init__(self, layer_data):
         try:
             self.operator = layer_data["fullname"]
             self.extra = layer_data["metadata_decoded"]["extra_fields"]
-            self.layer_number = self.extra["Layer Progressive Number"]["value"] # integer
-            self.target_elabid = self.extra["Target"]["value"] # elabid
-            self.laser_system_elabid = self.extra["Laser System"]["value"] # elabid
-            self.chamber_elabid = self.extra["Chamber"]["value"] # elabid
-            self.rheed_system_elabid = self.extra["RHEED System"]["value"] # elabid
+            self.layer_number = self.extra["Layer Progressive Number"][
+                "value"
+            ]  # integer
+            self.target_elabid = self.extra["Target"]["value"]  # elabid
+            self.laser_system_elabid = self.extra["Laser System"]["value"]  # elabid
+            self.chamber_elabid = self.extra["Chamber"]["value"]  # elabid
+            self.rheed_system_elabid = self.extra["RHEED System"]["value"]  # elabid
             self.deposition_time = self.extra["Duration"]["value"]
             self.deposition_time_unit = self.extra["Duration"]["unit"]
             self.repetition_rate = self.extra["Repetition rate"]["value"]
             self.repetition_rate_unit = self.extra["Repetition rate"]["unit"]
             try:
-                self.number_of_pulses = (float(self.deposition_time) * float(self.repetition_rate)).__floor__()
+                self.number_of_pulses = (
+                    float(self.deposition_time) * float(self.repetition_rate)
+                ).__floor__()
             except ValueError:
                 # Since number_of_pulses is required, if it can't be calculated raise error:
                 raise ValueError("""
@@ -33,22 +39,32 @@ class Layer:
     This has to be an error, since these fields are required by the NeXus standard.
     Please edit your eLabFTW entry and retry.
                 """)
-            self.temperature = self.extra["Heater temperature"]["value"] # Note: this field used to have a trailing space in its name
+            self.temperature = self.extra["Heater temperature"][
+                "value"
+            ]  # Note: this field used to have a trailing space in its name
             self.temperature_unit = self.extra["Heater temperature"]["unit"]
-            self.process_pressure = self.extra["Process pressure"]["value"] # Note: this field used to have a trailing space in its name
+            self.process_pressure = self.extra["Process pressure"][
+                "value"
+            ]  # Note: this field used to have a trailing space in its name
             self.process_pressure_unit = self.extra["Process pressure"]["unit"]
             self.heating_method = self.extra["Heating Method"]["value"]
             self.layer_thickness = self.extra["Thickness"]["value"]
             self.layer_thickness_unit = self.extra["Thickness"]["unit"]
             self.buffer_gas = self.extra["Buffer gas"]["value"]
             self.heater_target_distance = self.extra["Heater-target distance"]["value"]
-            self.heater_target_distance_unit = self.extra["Heater-target distance"]["unit"]
-            self.laser_fluence = self.extra["Laser Intensity"]["value"] # here fluence = intensity
+            self.heater_target_distance_unit = self.extra["Heater-target distance"][
+                "unit"
+            ]
+            self.laser_fluence = self.extra["Laser Intensity"][
+                "value"
+            ]  # here fluence = intensity
             self.laser_fluence_unit = "J/(s cm^2)"
             self.laser_spot_area = self.extra["Spot Area"]["value"]
             self.laser_spot_area_unit = "mm^2"
             try:
-                self.laser_energy = ( float(self.laser_fluence) * float(self.laser_spot_area) / 100 ).__round__(3)
+                self.laser_energy = (
+                    float(self.laser_fluence) * float(self.laser_spot_area) / 100
+                ).__round__(3)
                 self.laser_energy_unit = "J/s"
             except ValueError:
                 # Since laser_energy is NOT required, if it can't be calculated warn user but allow the software to continue execution:
@@ -61,39 +77,66 @@ class Layer:
                 self.laser_energy = "N/A"
                 self.laser_energy_unit = "J/s"
             # Laser rasternig section
-            self.laser_rastering_geometry = self.extra["Laser Rastering Geometry"]["value"]
-            self.laser_rastering_positions = self.extra["Laser Rastering Position"]["value"]
-            self.laser_rastering_velocities = self.extra["Laser Rastering Speed"]["value"]
+            self.laser_rastering_geometry = self.extra["Laser Rastering Geometry"][
+                "value"
+            ]
+            self.laser_rastering_positions = self.extra["Laser Rastering Position"][
+                "value"
+            ]
+            self.laser_rastering_velocities = self.extra["Laser Rastering Speed"][
+                "value"
+            ]
             # Pre annealing section
             self.pre_annealing_ambient_gas = self.extra["Buffer gas Pre"]["value"]
             self.pre_annealing_pressure = self.extra["Process pressure Pre"]["value"]
-            self.pre_annealing_temperature = self.extra["Heater temperature Pre"]["value"]
+            self.pre_annealing_temperature = self.extra["Heater temperature Pre"][
+                "value"
+            ]
             self.pre_annealing_duration = self.extra["Duration Pre"]["value"]
-            self.pre_annealing_pressure_unit = self.extra["Process pressure Pre"]["unit"]
-            self.pre_annealing_temperature_unit = self.extra["Heater temperature Pre"]["unit"]
+            self.pre_annealing_pressure_unit = self.extra["Process pressure Pre"][
+                "unit"
+            ]
+            self.pre_annealing_temperature_unit = self.extra["Heater temperature Pre"][
+                "unit"
+            ]
             self.pre_annealing_duration_unit = self.extra["Duration Pre"]["unit"]
             # Post annealing section
             self.post_annealing_ambient_gas = self.extra["Buffer gas PA"]["value"]
             self.post_annealing_pressure = self.extra["Process pressure PA"]["value"]
-            self.post_annealing_temperature = self.extra["Heater temperature PA"]["value"]
+            self.post_annealing_temperature = self.extra["Heater temperature PA"][
+                "value"
+            ]
             self.post_annealing_duration = self.extra["Duration PA"]["value"]
-            self.post_annealing_pressure_unit = self.extra["Process pressure PA"]["unit"]
-            self.post_annealing_temperature_unit = self.extra["Heater temperature PA"]["unit"]
+            self.post_annealing_pressure_unit = self.extra["Process pressure PA"][
+                "unit"
+            ]
+            self.post_annealing_temperature_unit = self.extra["Heater temperature PA"][
+                "unit"
+            ]
             self.post_annealing_duration_unit = self.extra["Duration PA"]["unit"]
 
             # Rejected but suggested by the NeXus standard:
-            #self.laser_rastering_coefficients = None
+            # self.laser_rastering_coefficients = None
         except KeyError as k:
             # Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
             # Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
-            raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the deposition layer entry on eLabFTW and make sure you used the correct Experiment template.")
+            raise KeyError(
+                f'The provided dictionary lacks a "{k}" key. Check the deposition layer entry on eLabFTW and make sure you used the correct Experiment template.'
+            )
         # Optional
         self.start_time = layer_data.get("created_at") or None
         self.description = layer_data.get("body") or None
+
     def get_instruments(self, apikey):
-        raw_lasersys_data = APIHandler(apikey).get_entry_from_elabid(self.laser_system_elabid, entryType="items")
-        raw_chamber_data = APIHandler(apikey).get_entry_from_elabid(self.chamber_elabid, entryType="items")
-        raw_rheedsys_data = APIHandler(apikey).get_entry_from_elabid(self.rheed_system_elabid, entryType="items")
+        raw_lasersys_data = APIHandler(apikey).get_entry_from_elabid(
+            self.laser_system_elabid, entryType="items"
+        )
+        raw_chamber_data = APIHandler(apikey).get_entry_from_elabid(
+            self.chamber_elabid, entryType="items"
+        )
+        raw_rheedsys_data = APIHandler(apikey).get_entry_from_elabid(
+            self.rheed_system_elabid, entryType="items"
+        )
         instruments_used = {
             "laser_system": raw_lasersys_data.get("title") or None,
             "deposition_chamber": raw_chamber_data.get("title") or None,
@@ -101,31 +144,50 @@ class Layer:
         }
         return instruments_used
 
+    # Three possible approaches for the next two methods: either return the raw data as retrieved from eLabFTW,
+    # or process it and return only the relevant information, or even simply return the list of filenames.
+    #
+    def fetch_textual_uploads(self, api_key):
+        """ """
+        return
+
+    def fetch_rheed_images(self, api_key):
+        return
+
+
 class Entrypoint:
-    '''
+    """
     Entrypoint(sample_data) - where sample_data is a Python dictionary.
 
     Meant to be used for eLabFTW Resources of the "Sample" category.
 
     The entrypoint is the starting point of the process of resolving the data chain.
     The entrypoint must be a dictionary containing the data of a sample, created directly from the JSON of the item endpoint on eLabFTW - which can be done through the function get_entry_from_elabid.
-    '''
+    """
+
     def __init__(self, sample_data):
         try:
             self.extra = sample_data["metadata_decoded"]["extra_fields"]
-            self.linked_items = sample_data["items_links"] # dict
-            self.batch_elabid = self.extra["Substrate batch"]["value"] # elabid
-            self.linked_experiments = sample_data["related_experiments_links"] # dict
-            self.linked_experiments_elabid = [ i["entityid"] for i in self.linked_experiments ] # list of elabid
+            self.linked_items = sample_data["items_links"]  # dict
+            self.batch_elabid = self.extra["Substrate batch"]["value"]  # elabid
+            self.linked_experiments = sample_data["related_experiments_links"]  # dict
+            self.linked_experiments_elabid = [
+                i["entityid"] for i in self.linked_experiments
+            ]  # list of elabid
         except KeyError as k:
             # Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
             # Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
-            raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the sample entry on eLabFTW and make sure you used the correct Resource template.")
+            raise KeyError(
+                f'The provided dictionary lacks a "{k}" key. Check the sample entry on eLabFTW and make sure you used the correct Resource template.'
+            )
         # Non-required attributes:
-        self.name = sample_data.get("title") or None # error prevention is more important than preventing empty fields here
+        self.name = (
+            sample_data.get("title") or None
+        )  # error prevention is more important than preventing empty fields here
+
 
 class Material:
-    '''
+    """
     Material(material_data) - where material_data is a Python dictionary.
 
     Meant to be used for eLabFTW Resources of either the "PLD Target" or the "Substrate" categories.
@@ -134,13 +196,16 @@ class Material:
         * Name and formula;
         * Shape and dimensions;
         * Misc.
-    '''
+    """
+
     def __init__(self, material_data):
         try:
-            self.name = material_data["title"] # required
+            self.name = material_data["title"]  # required
             self.extra = material_data["metadata_decoded"]["extra_fields"]
             self.compound_elabid = self.extra["Compound"]["value"]
-            self.dimensions = str(self.extra["Size"]["value"]) # strings have a .count() method
+            self.dimensions = str(
+                self.extra["Size"]["value"]
+            )  # strings have a .count() method
             if self.dimensions.count("mm") == 2:
                 self.dimensions_unit = "mm x mm"
             elif self.dimensions[-1] == '"':
@@ -150,23 +215,30 @@ class Material:
         except KeyError as k:
             # Some keys are not required and can be called through the .get() method - which is permissive and allows null values;
             # Other keys are required so if they can't be called (invalid or null) raise error and stop execution of the program:
-            raise KeyError(f"The provided dictionary lacks a \"{k}\" key. Check the target/substrate entry on eLabFTW and make sure you used the correct Resource template.")
+            raise KeyError(
+                f'The provided dictionary lacks a "{k}" key. Check the target/substrate entry on eLabFTW and make sure you used the correct Resource template.'
+            )
+
     def get_compound_data(self, apikey):
-        raw_compound_data = APIHandler(apikey).get_entry_from_elabid(self.compound_elabid, entryType="items")
+        raw_compound_data = APIHandler(apikey).get_entry_from_elabid(
+            self.compound_elabid, entryType="items"
+        )
         name = raw_compound_data["title"]
         extra = raw_compound_data["metadata_decoded"]["extra_fields"]
         formula = extra.get("Chemical formula")
-        cas = extra.get("CAS number ") or { "value": None }
+        cas = extra.get("CAS number ") or {"value": None}
         compound_data = {
-            "name" : name,
-            "chemical_formula" : formula.get("value"),
-            "cas_number" : cas.get("value")
+            "name": name,
+            "chemical_formula": formula.get("value"),
+            "cas_number": cas.get("value"),
         }
         return compound_data
+
     def get_compound_formula(self, apikey):
         formula = self.get_compound_data(apikey).get("chemical_formula")
         return formula
 
+
 class Substrate(Material):
     def __init__(self, material_data):
         super().__init__(material_data)
@@ -176,13 +248,16 @@ class Substrate(Material):
             self.miscut_angle_unit = self.extra["Miscut Angle"]["unit"]
             self.miscut_direction = self.extra["Miscut Direction"]["value"]
             # Not present (yet) on eLabFTW for Substrates:
-            self.thickness = "" #self.extra["Thickness"]["value"]
-            self.thickness_unit = "μm" #self.extra["Thickness"]["unit"]
+            self.thickness = ""  # self.extra["Thickness"]["value"]
+            self.thickness_unit = "μm"  # self.extra["Thickness"]["unit"]
             self.surface_treatment = self.extra["Surface treatment"]["value"]
             self.manufacturer = self.extra["Supplier"]["value"]
             self.batch_id = self.extra["Batch ID"]["value"]
         except KeyError as k:
-            raise KeyError(f"The provided dictionary lacks a \"{k}\" key - which is specific for substrates. Check the {self.name} substrate entry on eLabFTW and make sure you used the correct Resource template.")
+            raise KeyError(
+                f'The provided dictionary lacks a "{k}" key - which is specific for substrates. Check the {self.name} substrate entry on eLabFTW and make sure you used the correct Resource template.'
+            )
+
 
 class Target(Material):
     def __init__(self, material_data):
@@ -194,13 +269,15 @@ class Target(Material):
             self.solid_form = self.extra["Solid form"]["value"]
             self.manufacturer = self.extra["Supplier"]["value"]
         except KeyError as k:
-            raise KeyError(f"The provided dictionary lacks a \"{k}\" key - which is specific for PLD targets. Check the {self.name} target entry on eLabFTW and make sure you used the correct Resource template.")
+            raise KeyError(
+                f'The provided dictionary lacks a "{k}" key - which is specific for PLD targets. Check the {self.name} target entry on eLabFTW and make sure you used the correct Resource template.'
+            )
         # Non-required attributes:
         self.description = material_data.get("body") or ""
 
 
-
-if __name__=="__main__":
-    head = Header("MyApiKey-123456789abcdef")
+if __name__ == "__main__":
+    head = APIHandler("MyApiKey-123456789abcdef")
     print(f"Example header:\n\t{head.header}\n")
     print("Warning: you're not supposed to be running this as the main program.")
+