diff --git a/apiserver/storage/JsonFileStorageAdapter.py b/apiserver/storage/JsonFileStorageAdapter.py
index 40162d59ede9b9fcff9a6b92913e160c1f0f4b9b..8d36b0bd003f958b7128967bcf53f83d91842697 100644
--- a/apiserver/storage/JsonFileStorageAdapter.py
+++ b/apiserver/storage/JsonFileStorageAdapter.py
@@ -8,15 +8,17 @@ from apiserver.config import ApiserverSettings
from .LocationStorage import (AbstractLocationDataStorageAdapter, LocationData,
LocationDataType)
+#from dataclasses import dataclass, asdict
+#perhaps even that:
+#from dataclasses_json import dataclass_json
-class StoredData:
+from pydantic import BaseModel
+
+#@dataclass
+class StoredData(BaseModel):
actualData: LocationData
users: List[str]
- def toDict(self):
- return {'actualData': self.actualData.__dict__, 'users': self.users}
-
-
class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
""" This stores LocationData via the StoredData Object as json files
@@ -35,14 +37,24 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
if not (os.path.exists(self.data_dir) and os.path.isdir(self.data_dir)):
raise Exception(f"Data Directory {self.data_dir} does not exist.")
- def __setup_path(self, value):
+ def __setup_path(self, value: str) -> str:
localpath = os.path.join(self.data_dir, value)
if not (os.path.isdir(localpath)):
- # This type has apparently not yet been used at all,
- # create its directory and return an empty json file
os.mkdir(localpath)
return localpath
+ def __get_object_path(self, value: str, oid: str) -> str:
+ localpath = os.path.join(self.data_dir, value)
+ fullpath = os.path.join(localpath, oid)
+ if not os.path.isfile(fullpath):
+ raise FileNotFoundError(f"The requested object ({oid}) does not exist.")
+ return fullpath
+
+ def __get_unique_id(self, path: str) -> str:
+ oid = str(uuid.uuid4())
+ while (os.path.exists(os.path.join(path, oid))):
+ oid = str(uuid.uuid4())
+ return oid
def get_list(self, n_type: LocationDataType) -> List:
local_path = self.__setup_path(n_type.value)
@@ -52,59 +64,46 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
# and the LocationData name (which is inside the json)
retList = []
for f in allFiles:
- with open(os.path.join(local_path, f)) as file:
- data = json.load(file)
- retList.append({data['actualData']['name']: f})
+ data = StoredData.parse_file(os.path.join(local_path, f))
+ #with open(os.path.join(local_path, f)) as file:
+ # data = json.load(file)
+ retList.append({data.actualData.name: f})
return retList
def add_new(self, n_type: LocationDataType, data: LocationData, usr: str):
localpath = self.__setup_path(value=n_type.value)
- # create a unique oid, by randomly generating one,
- # and re-choosing if it is already taken
- oid = str(uuid.uuid4())
- while (os.path.exists(os.path.join(localpath, oid))):
- oid = str(uuid.uuid4())
- toStore = StoredData()
- toStore.users = [usr]
- toStore.actualData = data
+ oid = self.__get_unique_id(path=localpath)
+ toStore = StoredData(users=[usr], actualData=data)
with open(os.path.join(localpath, oid), 'w') as json_file:
- json.dump(toStore.toDict(), json_file)
- return {oid: data}
-
- def get_details(self, type: LocationDataType, oid: str):
- localpath = os.path.join(self.data_dir, type.value)
- fullpath = os.path.join(localpath, oid)
- if not os.path.isfile(fullpath):
- raise FileNotFoundError(f"The requested object ({oid}) does not exist.")
- with open(fullpath) as file:
- data = json.load(file)
- return data['actualData']
+ json.dump(toStore.dict(), json_file)
+ return (oid, data)
- def update_details(self, type: LocationDataType, oid: str, data: LocationData, usr: str):
- localpath = os.path.join(self.data_dir, type.value)
- fullpath = os.path.join(localpath, oid)
- if not os.path.isfile(fullpath):
- raise FileNotFoundError(f"The requested object ({oid}) does not exist.")
+ def __load_object(self, path):
+ # move to data class?
+ with open(path, 'r') as f:
+ data = json.load(f)
+ return data
- toStore = StoredData()
- toStore.actualData = data
- # get permissions from old file
- with open(fullpath) as file:
- old_data = json.load(file)
- toStore.users = old_data['users']
+ def get_details(self, n_type: LocationDataType, oid: str):
+ full_path = self.__get_object_path(value=n_type.value, oid=oid)
+ obj = self.__load_object(path=full_path)
+ return obj['actualData']
+
- with open(fullpath, 'w') as file:
- json.dump(toStore.toDict(), file)
- return {oid: data}
+ def update_details(self, n_type: LocationDataType, oid: str, data: LocationData, usr: str):
+ full_path = self.__get_object_path(value=n_type.value, oid=oid)
+ old_data = self.__load_object(path=full_path)
+ old_data['actualData']=data
- def delete(self, type: LocationDataType, oid: str, usr: str):
- localpath = os.path.join(self.data_dir, type.value)
- fullpath = os.path.join(localpath, oid)
+ full_path = self.__get_object_path(value=n_type.value, oid=oid)
+ with open(full_path, 'w') as file:
+ json.dump(old_data.json(), file)
- if not os.path.isfile(fullpath):
- raise FileNotFoundError(f"The requested object {oid} does not exist.")
+ return (oid, data)
+ def delete(self, n_type: LocationDataType, oid: str, usr: str):
+ fullpath = self.__get_object_path(value=n_type.value, oid=oid)
os.remove(fullpath)
def get_owner(self, type: LocationDataType, oid: str):
diff --git a/tests/storage_tests/test_jsonbackend.py b/tests/storage_tests/test_jsonbackend.py
index c27ac1cde58c9eb86325b5abb3c91048e89ed906..b0c8cbf1ec5e15705a29942fe4a7417fc6e95f30 100644
--- a/tests/storage_tests/test_jsonbackend.py
+++ b/tests/storage_tests/test_jsonbackend.py
@@ -1,7 +1,7 @@
import unittest
-from apiserver.storage.JsonFileStorageAdapter import JsonFileStorageAdapter
-from apiserver.storage import LocationDataType
+from apiserver.storage.JsonFileStorageAdapter import JsonFileStorageAdapter, StoredData
+from apiserver.storage import LocationDataType, LocationData
from collections import namedtuple
import os
import pathlib
@@ -31,9 +31,27 @@ class SomeTests(unittest.TestCase):
test_config = Settings('/tmp/json_test/blah/')
self.assertRaises(Exception, JsonFileStorageAdapter, test_config)
-
+ def test_add_new(self):
+ d = LocationData(name='bla', url='local')
+ (oid, data) = self.store.add_new(n_type=LocationDataType.DATASET, data=d, usr='test_user')
+ self.assertEquals(d, data, "Data should be equal")
+ self.assertIsNotNone(oid)
+
+ def test_add_and_read(self):
+ l_data = LocationData(name='test1', url='http://n.go', metadata=[])
+ (oid, data) = self.store.add_new(n_type=LocationDataType.DATASET, data=l_data, usr='test_user')
+ self.assertEquals(l_data, data, "Data should be equal")
+ self.assertIsNotNone(oid)
+ print(data)
+
+ lst = self.store.get_list(n_type=LocationDataType.DATASET)
+ self.assertEqual(len(lst), 1, 'One should be there')
+
+
+
+
+
+
-
-
\ No newline at end of file