Skip to content
Snippets Groups Projects
Commit a5e89c04 authored by Christian Boettcher's avatar Christian Boettcher
Browse files

doing some renaming and restructuring for easier automated testing

parent 58490616
No related branches found
No related tags found
1 merge request!1apiserver based on fastAPI
**__pycache__/
**app/
\ No newline at end of file
__pycache__/
app/
\ No newline at end of file
from pydantic import BaseModel
from typing import Optional
from typing import Dict
from enum import Enum
class LocationDataType(Enum):
DATASET: str = 'dataset'
STORAGETARGET: str = 'storage_target'
class LocationData(BaseModel):
name: str
url: str
metadata: Optional[Dict[str, str]]
class AbstractLocationDataStorageAdapter:
def getList(self, type: LocationDataType):
raise NotImplementedError()
def addNew(self, type: LocationDataType, data: LocationData):
raise NotImplementedError()
def getDetails(self, type: LocationDataType, id: str):
raise NotImplementedError()
def updateDetails(self, type:LocationDataType, id:str, data: LocationData):
raise NotImplementedError()
...@@ -2,8 +2,12 @@ FROM tiangolo/uvicorn-gunicorn:python3.8 ...@@ -2,8 +2,12 @@ FROM tiangolo/uvicorn-gunicorn:python3.8
LABEL maintainer="Christian Böttcher <c.boettcher@fz-juelich.de>" LABEL maintainer="Christian Böttcher <c.boettcher@fz-juelich.de>"
RUN mkdir -p /app/data
RUN apt update && apt upgrade -y RUN apt update && apt upgrade -y
RUN pip install --no-cache-dir fastapi RUN pip install --no-cache-dir fastapi
COPY ./api-server.py /app/main.py COPY ./main.py /app/
\ No newline at end of file COPY ./LocationStorage.py /app/
COPY ./JsonFileStorageAdapter.py /app/
\ No newline at end of file
...@@ -4,9 +4,19 @@ import uuid ...@@ -4,9 +4,19 @@ import uuid
from LocationStorage import AbstractLocationDataStorageAdapter, LocationData, LocationDataType from LocationStorage import AbstractLocationDataStorageAdapter, LocationData, LocationDataType
from typing import List
DEFAULT_JSON_FILEPATH: str = "./app/data" DEFAULT_JSON_FILEPATH: str = "./app/data"
class StoredData:
actualData: LocationData
users: List[str]
# This stores LocationData via the StoredData Object as json files
# These Jsonfiles then contain the actualData, as well as the users with permissions for this LocationData
# all users have full permission to to anything with this dataobject, uncluding removing their own access (this might trigger a confirmation via the frontend, but this is not enforced via the api)
# IMPORTANT: The adapter does not check for authentication or authorization, it should only be invoked if the permissions have been checked
class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter): class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
data_dir: str data_dir: str
...@@ -29,10 +39,10 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter): ...@@ -29,10 +39,10 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
for f in allFiles: for f in allFiles:
with open(os.path.join(localpath, f)) as file: with open(os.path.join(localpath, f)) as file:
data = json.load(file) data = json.load(file)
retList.append({data['name'] : f}) retList.append({data['actualData']['name'] : f})
return retList return retList
def addNew(self, type: LocationDataType, data: LocationData): def addNew(self, type: LocationDataType, data: LocationData, usr: str):
localpath = os.path.join(self.data_dir, type.value) localpath = os.path.join(self.data_dir, type.value)
if not (os.path.isdir(localpath)): if not (os.path.isdir(localpath)):
# This type has apparently not yet been used at all, therefore we need to create its directory # This type has apparently not yet been used at all, therefore we need to create its directory
...@@ -41,8 +51,11 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter): ...@@ -41,8 +51,11 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
id = str(uuid.uuid4()) id = str(uuid.uuid4())
while (os.path.exists(os.path.join(localpath, id))): while (os.path.exists(os.path.join(localpath, id))):
id = str(uuid.uuid4()) id = str(uuid.uuid4())
toStore = StoredData()
toStore.users = [usr]
toStore.actualData = data
with open(os.path.join(localpath, id), 'w') as json_file: with open(os.path.join(localpath, id), 'w') as json_file:
json.dump(data.__dict__, json_file) json.dump(toStore.__dict__, json_file)
return {id : data} return {id : data}
def getDetails(self, type: LocationDataType, id: str): def getDetails(self, type: LocationDataType, id: str):
...@@ -52,13 +65,34 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter): ...@@ -52,13 +65,34 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
raise FileNotFoundError('The requested Object does not exist.') raise FileNotFoundError('The requested Object does not exist.')
with open(fullpath) as file: with open(fullpath) as file:
data = json.load(file) data = json.load(file)
return data return data['actualData']
def updateDetails(self, type:LocationDataType, id:str, data: LocationData): def updateDetails(self, type:LocationDataType, id:str, data: LocationData, usr: str):
localpath = os.path.join(self.data_dir, type.value) localpath = os.path.join(self.data_dir, type.value)
fullpath = os.path.join(localpath, id) fullpath = os.path.join(localpath, id)
if not os.path.isfile(fullpath): if not os.path.isfile(fullpath):
raise FileNotFoundError('The requested Object does not exist.') raise FileNotFoundError('The requested Object does not exist.')
toStore = StoredData()
# get permissions from old file
with open(fullpath) as file:
data = json.load(file)
toStore.users = data['users']
toStore.actualData = data
with open(fullpath, 'w') as file: with open(fullpath, 'w') as file:
json.dump(data.__dict__, file) json.dump(data.__dict__, file)
return {id : data} return {id : data}
def getOwner(self, type: LocationDataType(), id: str):
raise NotImplementedError()
def checkPerm(self, type: LocationDataType, id: str, usr: str):
raise NotImplementedError()
def addPerm(self, type: LocationDataType, id: str, authUsr: str, newUser: str):
raise NotImplementedError()
def rmPerm(self, type: LocationDataType, id: str, usr: str, rmUser: str):
raise NotImplementedError()
\ No newline at end of file
from pydantic import BaseModel
from typing import Optional
from typing import Dict
from enum import Enum
class LocationDataType(Enum):
DATASET: str = 'dataset'
STORAGETARGET: str = 'storage_target'
class LocationData(BaseModel):
name: str
url: str
metadata: Optional[Dict[str, str]]
#
'''
This is an abstract storage adapter for storing information about datasets, storage targets and similar things.
It can easily be expanded to also store other data (that has roughly similar metadata), just by expanding the LocationDataType Enum.
In general, all data is public. This means, that the adapter does not do any permission checking, except when explicitly asked via the checkPerm function.
The caller therefore has to manually decide when to check for permissions, and not call any function unless it is already authorized (or does not need any authorization).
The usr: str (the user id) that is required for several functions, is a unique and immutable string, that identifies the user. This can be a verified email or a user name.
The management of authentication etc. is 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.
'''
class AbstractLocationDataStorageAdapter:
# get a list of all LocationData Elements with the provided type, as pairs of {name : id}
def getList(self, type: LocationDataType):
raise NotImplementedError()
# add a new element of the provided type, assign and return the id and the new data as {id : LocationData}
def addNew(self, type: LocationDataType, data: LocationData, usr: str):
raise NotImplementedError()
# return the LocationData of the requested object (identified by id and type)
def getDetails(self, type: LocationDataType, id: str):
raise NotImplementedError()
# change the details of the requested object, return {id : newData}
def updateDetails(self, type:LocationDataType, id:str, data: LocationData, usr: str):
raise NotImplementedError()
# return the owner of the requested object; if multiple owners are set, return them is a list
def getOwner(self, type: LocationDataType, id: str):
raise NotImplementedError()
# check if the given user has permission to change the given object
def checkPerm(self, type: LocationDataType, id: str, usr: str):
raise NotImplementedError()
# add user to file perm
def addPerm(self, type: LocationDataType, id: str, usr: str):
raise NotImplementedError()
# remove user from file perm
def rmPerm(self, type: LocationDataType, id: str, usr: str):
raise NotImplementedError()
...@@ -23,7 +23,7 @@ pip install uvicorn[standard] ...@@ -23,7 +23,7 @@ pip install uvicorn[standard]
To start the server, run To start the server, run
```bash ```bash
uvicorn api-server:app --reload uvicorn main:app --reload
``` ```
Withour any other options, this starts your server on `<localhost:8000>`. Withour any other options, this starts your server on `<localhost:8000>`.
......
File moved
fastapi==0.63.0
pytest==6.2.4
requests==2.25.1
uvicorn==0.13.4
\ No newline at end of file
# These Tests only check if every api path that should work is responding to requests, the functionality is not yet checked
# Therefore this only detects grievous errors in the request handling.
from fastapi.testclient import TestClient
from apiserver.LocationStorage import LocationDataType
from apiserver import main
client = TestClient(main.app)
def test_root():
rsp = client.get('/')
assert rsp.status_code >= 200 and rsp.status_code < 300 # any 200 response is fine, as a get to the root should not return any error
def test_types():
for location_type in LocationDataType:
rsp = client.get('/' + location_type.value)
assert rsp.status_code >= 200 and rsp.status_code < 300 # any 200 response is fine, as a get to the datatypes should not return any error
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment