Skip to content
Snippets Groups Projects
Commit b1e18ee1 authored by Jedrzej Rybicki's avatar Jedrzej Rybicki
Browse files

removing type (reserverd keyword)

parent e159657b
No related branches found
No related tags found
No related merge requests found
import os
import json
import os
import uuid
from .LocationStorage import AbstractLocationDataStorageAdapter, LocationData, LocationDataType
from typing import List
from apiserver.config import ApiserverSettings
from typing import List
from .LocationStorage import (AbstractLocationDataStorageAdapter, LocationData,
LocationDataType)
class StoredData:
......@@ -35,31 +35,30 @@ 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 get_list(self, type: LocationDataType) -> List:
localpath = os.path.join(self.data_dir, type.value)
def __setup_path(self, value):
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 []
else:
allFiles = [f for f in os.listdir(
localpath) if os.path.isfile(os.path.join(localpath, f))]
# now each file has to be checked for its filename (= oid)
# and the LocationData name (which is inside the json)
retList = []
for f in allFiles:
with open(os.path.join(localpath, f)) as file:
data = json.load(file)
retList.append({data['actualData']['name']: f})
return retList
def add_new(self, type: LocationDataType, data: LocationData, usr: str):
localpath = os.path.join(self.data_dir, type.value)
if not (os.path.isdir(localpath)):
# This type has apparently not yet been used at all,
# therefore we need to create its directory
os.mkdir(localpath)
return localpath
def get_list(self, n_type: LocationDataType) -> List:
local_path = self.__setup_path(n_type.value)
allFiles = [f for f in os.listdir(
local_path) if os.path.isfile(os.path.join(local_path, f))]
# now each file has to be checked for its filename (= oid)
# 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})
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())
......
from pydantic import BaseModel
from typing import Optional
from typing import Dict
from enum import Enum
from typing import Dict, Optional, List
from pydantic import BaseModel
class LocationDataType(Enum):
......@@ -36,38 +35,40 @@ class AbstractLocationDataStorageAdapter:
done by the caller, this adapter assumes that the user id fulfills the criteria.
Permissions are stored as a list of user ids, and every id is authorized for full access.
'''
# get a list of all LocationData Elements with the provided type, as pairs of {name : id}
def get_list(self, type: LocationDataType):
def get_list(self, n_type: LocationDataType) -> List:
# get a list of all LocationData Elements with the provided type, as pairs of {name : id}
raise NotImplementedError()
# add a new element of the provided type, assign and return the id and the new data as {id : LocationData}
def add_new(self, type: LocationDataType, data: LocationData, usr: str):
def add_new(self, n_type: LocationDataType, data: LocationData, usr: str):
# add a new element of the provided type, assign and return the id and
# the new data as {id : LocationData}
raise NotImplementedError()
# return the LocationData of the requested object (identified by oid and type)
def get_details(self, type: LocationDataType, oid: str):
def get_details(self, n_type: LocationDataType, oid: str):
# return the LocationData of the requested object (identified by oid and type)
raise NotImplementedError()
# change the details of the requested object, return {oid : newData}
def update_details(self, type: LocationDataType, oid: str, data: LocationData, usr: str):
def update_details(self, n_type: LocationDataType, oid: str, data: LocationData, usr: str):
# change the details of the requested object, return {oid : newData}
raise NotImplementedError()
def delete(self, type: LocationDataType, oid: str, usr: str):
def delete(self, n_type: LocationDataType, oid: str, usr: str):
raise NotImplementedError()
# return the owner of the requested object; if multiple owners are set, return them is a list
def get_owner(self, type: LocationDataType, oid: str):
def get_owner(self, n_type: LocationDataType, oid: str):
# return the owner of the requested object; if multiple owners are set,
# return them is a list
raise NotImplementedError()
# check if the given user has permission to change the given object
def check_perm(self, type: LocationDataType, oid: str, usr: str):
def check_perm(self, n_type: LocationDataType, oid: str, usr: str):
# check if the given user has permission to change the given object
raise NotImplementedError()
# add user to file perm
def add_perm(self, type: LocationDataType, oid: str, usr: str):
def add_perm(self, n_type: LocationDataType, oid: str, usr: str):
# add user to file perm
raise NotImplementedError()
# remove user from file perm
def rm_perm(self, type: LocationDataType, oid: str, usr: str):
def rm_perm(self, n_type: LocationDataType, oid: str, usr: str):
# remove user from file perm
raise NotImplementedError()
......@@ -10,7 +10,6 @@ import shutil
class SomeTests(unittest.TestCase):
def setUp(self):
Settings = namedtuple('Settings',['json_storage_path'])
self.test_config = Settings('/tmp/json_test/')
pathlib.Path(self.test_config.json_storage_path).mkdir(parents=True, exist_ok=True)
......@@ -19,11 +18,22 @@ class SomeTests(unittest.TestCase):
def tearDown(self):
if os.path.exists(self.test_config.json_storage_path):
print('Paht exists. Removing')
print('Path exists. Removing')
shutil.rmtree(self.test_config.json_storage_path)
def test_getList(self):
def test_get_emptyList(self):
test_type = LocationDataType.DATASET
lst = self.store.get_list(type=test_type)
lst = self.store.get_list(n_type=test_type)
self.assertEqual(lst, [], 'Id should not be none')
def test_not_path(self):
Settings = namedtuple('Settings',['json_storage_path'])
test_config = Settings('/tmp/json_test/blah/')
self.assertRaises(Exception, JsonFileStorageAdapter, test_config)
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment