From 7245c757c3f33ab4607533953c435e72ab8920b4dd94e0550b9c1241698e32cf Mon Sep 17 00:00:00 2001 From: PioApocalypse Date: Wed, 28 Jan 2026 15:41:20 +0100 Subject: [PATCH] reworks chained_requests classes for parsing, functions for requests also in general I shouldn't put error handling inside a class --- src/chained_requests.py | 89 +++++++++++++++++++++++++++++------------ 1 file changed, 63 insertions(+), 26 deletions(-) diff --git a/src/chained_requests.py b/src/chained_requests.py index bc9f9de..3056d66 100644 --- a/src/chained_requests.py +++ b/src/chained_requests.py @@ -2,42 +2,79 @@ import os, json, requests from getpass import getpass from classes import Header -''' -Starting from the sample's page we'll use this code to pull and return all data from the entries related to the sample and its fabrication. -''' +class Layers: + def __init__(self, layer_data): + try: + self.extra = layer_data["metadata_decoded"]["extra_fields"] + except KeyError as k: + raise KeyError(f"The provided dictionary lacks a \"{k}\" key.") + self.target_elabid = self.extra["Target"]["value"] + self.start_time = layer_data.get("created_at") + self.operator = layer_data.get("fullname") + self.description = layer_data.get("body") + self.deposition_time = self.extra["Duration"]["value"] + self.repetition_rate = self.extra["Repetition rate"]["value"] + self.number_of_pulses = float(self.deposition_time) * float(self.repetition_rate) + self.temperature = self.extra["Heater temperature "]["value"] + self.heating_method = self.extra["Heating Method"]["value"] class Entrypoint: ''' - Use: Entrypoint(elabid) + Entrypoint(sample_data) - where sample_data is a Python dictionary. - The entrypoint is the starting point of the process of resolving the data chain. The entrypoint must be a sample. - - Currently, the elabid of the sample must be provided. In the future a different method will eventually be implemented. + 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, elabid): + def __init__(self, sample_data): ''' - Constructor method. - self.sample_data is a dictionary of all data on the sample's eLab entry. - self.linked_experiments is a sub-dictionary which only includes links to related experiments. - self.batch_elabid is an integer (or NoneType if there's an error) containing the elabid of the associated substrate batch entry. + Attributes: + - self.linked_experiments is a sub-dictionary which only includes links to related experiments. + - self.linked_items is a sub-dictionary which only includes links to related items. + - self.batch_elabid is an integer (or NoneType if there's an error) containing the elabid of the associated substrate batch entry. ''' - header = Header(apikey).dump - self.sample_data = requests.get( - headers = header, - url = f"https://elabftw.fisica.unina.it/api/v2/items/{elabid}", - verify=True - ).json() - self.linked_experiments = self.sample_data.get("related_experiments_links") or None try: - self.batch_elabid = self.sample_data["metadata_decoded"]["extra_fields"]["Substrate batch"]["value"] - except KeyError as k: # if no metadata exists - which means there's a problem with the entry - self.batch_elabid = None + self.extra = sample_data["metadata_decoded"]["extra_fields"] + except KeyError as k: + raise KeyError(f"The provided dictionary lacks a \"{k}\" key.") + self.linked_experiments = sample_data.get("related_experiments_links") or None + self.linked_items = sample_data.get("items_links") or None + self.batch_elabid = self.extra["Substrate batch"]["value"] + + +def get_entry_from_elabid(elabid, entryType="items"): + ''' + Function which returns entrypoint data (as dictionary) from its elabid. + ''' + header = Header(apikey).dump + response = requests.get( + headers = header, + url = f"{ELABFTW_API_URL}/{entryType}/{elabid}", + verify=True + ) + if response.status_code // 100 in [2,3]: + entry_data = response.json() + return entry_data + else: + raise ConnectionError(f"HTTP request failed with status code: {response.status_code}.") if __name__=="__main__": print("===== DEBUG MODE! =====") + ELABFTW_API_URL = "https://elabftw.fisica.unina.it/api/v2" apikey = getpass("Paste API key here: ") + entryType = None + while entryType not in ["items", "experiments"]: + eT = input("Enter a valid entry type [items, experiments]: ") + match eT: + case "items" | "i" | "item": + entryType = "items" + case "experiments" | "e" | "exp" | "experiment": + entryType = "experiments" + case _: + pass elabid = input("Input elabid here [default = 1108]: ") or 1108 - chain = Entrypoint(elabid) - print(json.dumps(chain.sample_data)) - print(json.dumps(chain.linked_experiments)) - print(chain.batch_elabid) \ No newline at end of file + data = get_entry_from_elabid(elabid, entryType) + if entryType == "experiments": + layer = Layers(data) + print(layer.__dict__) + # print(json.dumps(chain.sample_data)) + # print(json.dumps(chain.linked_experiments)) + # print(chain.batch_elabid) \ No newline at end of file