From 9ece85944a4effb8a4f6b90d251cbab8c012cd7c Mon Sep 17 00:00:00 2001 From: "Prof. Dr. Matthias Jung DL9MJ" Date: Fri, 16 May 2025 00:20:41 +0200 Subject: [PATCH] feature: add directus api --- api/directus.py | 52 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 api/directus.py diff --git a/api/directus.py b/api/directus.py new file mode 100644 index 0000000..08cc09d --- /dev/null +++ b/api/directus.py @@ -0,0 +1,52 @@ +import json + +import requests + + +class DirectusAPI: + def __init__(self, config): + self.config = config + + def __get_raw(self, endpoint, params = None): + if params is None: + params = {} + for k in params.keys(): + if type(params[k]) is dict: + params[k] = json.dumps(params[k]) + url = self.config.base_url + endpoint + result = requests.get( + url, + params = params, + headers = { + "Authorization": f"Bearer {self.config.access_token}", + "Cache-Control": "no-store" + } + ) + return result + + def get(self, endpoint, params = None): + if params is None: + params = {} + result = self.__get_raw(endpoint, params) + try: + result_json = result.json() + if "errors" in result_json.keys(): + print(result_json["errors"]) + raise Exception("Error communicating with API") + data = result_json["data"] + except Exception as e: + print(f"Error: {e}") + return None + return data + + def get_one(self, endpoint, params = None): + if params is None: + params = {} + results = self.get(endpoint, params) + if len(results) == 1: + return results[0] + elif len(results) == 0: + return None + else: + raise Exception("One result expected, but got more than one result") + \ No newline at end of file