Skip to content
Snippets Groups Projects
Commit ad45402f authored by Robert Goldmann's avatar Robert Goldmann
Browse files

externalized measurement routes

parent cb8d8639
Branches
Tags
No related merge requests found
......@@ -3,7 +3,7 @@ import os
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger
from TheCodeLabs_FlaskUtils.FlaskBaseApp import FlaskBaseApp
from blueprints import Routes, Devices
from blueprints import Routes, Devices, Sensors, Measurements
from logic import Constants
LOGGER = DefaultLogger().create_logger_if_not_exists(Constants.APP_NAME)
......@@ -16,7 +16,8 @@ class StorageLeaf(FlaskBaseApp):
def _register_blueprints(self, app):
app.register_blueprint(Routes.construct_blueprint(self._settings, self._version))
app.register_blueprint(Devices.construct_blueprint(self._settings))
app.register_blueprint(Sensors.construct_blueprint(self._settings))
app.register_blueprint(Measurements.construct_blueprint(self._settings))
return app
......
from enum import Enum
from flask import Blueprint, jsonify
from logic.Database import Database
class DeviceParameters(Enum):
DEVICE = 'device'
SENSORS = 'sensors'
@staticmethod
def get_values():
return [m.value for m in DeviceParameters]
def construct_blueprint(settings):
devices = Blueprint('devices', __name__)
......
from typing import Dict
from flask import Blueprint, jsonify, request
from logic.AuthenticationWrapper import require_api_key
from logic.Database import Database
from logic.Parameters import DeviceParameters, SensorParameters
from logic.RequestValidator import RequestValidator, ValidationError
def construct_blueprint(settings):
measurements = Blueprint('measurements', __name__)
@measurements.route('/measurements', methods=['GET'])
def get_all_measurements():
database = Database(settings['database']['databasePath'])
return jsonify(database.measurementAccess.get_all_measurements())
@measurements.route('/measurement/<int:measurementID>', methods=['GET'])
def get_measurement(measurementID):
database = Database(settings['database']['databasePath'])
return jsonify(database.measurementAccess.get_measurement(measurementID))
@measurements.route('/measurements', methods=['POST'])
@require_api_key(password=settings['api']['key'])
def addMeasurement():
try:
parameters = RequestValidator.validate(request, DeviceParameters.get_values())
database = Database(settings['database']['databasePath'])
deviceName = parameters[DeviceParameters.DEVICE.value]
if not database.deviceAccess.get_device_by_name(deviceName):
database.deviceAccess.add_device(deviceName)
device = database.deviceAccess.get_device_by_name(deviceName)
sensors = parameters[DeviceParameters.SENSORS.value]
for sensor in sensors:
sensorParams = RequestValidator.validate_parameters(sensor,
SensorParameters.get_values(),
f'sensor "{sensor}"')
sensor = __add_sensor_if_not_exists(database, int(device['id']), sensorParams)
database.measurementAccess.add_measurement(int(sensor['id']),
sensorParams[SensorParameters.VALUE.value])
except ValidationError as e:
return e.response, 400
return jsonify({'success': True})
def __add_sensor_if_not_exists(database: Database, deviceID: int, sensorParams: Dict) -> Dict[str, str]:
sensorName = sensorParams[SensorParameters.NAME.value]
sensorType = sensorParams[SensorParameters.TYPE.value]
sensor = database.sensorAccess.get_sensor_by_name_and_device_id(deviceID, sensorName)
if sensor:
return sensor
database.sensorAccess.add_sensor(deviceID, sensorName, sensorType)
return database.sensorAccess.get_sensor_by_name_and_device_id(deviceID, sensorName)
return measurements
import json
import os
from enum import Enum
from typing import Dict
import yaml
from flask import Blueprint, request, jsonify, render_template
from flask import Blueprint, render_template
from logic import Constants
from logic.AuthenticationWrapper import require_api_key
from logic.Database import Database
from logic.RequestValidator import ValidationError, RequestValidator
def construct_blueprint(settings, version):
......@@ -29,49 +24,4 @@ def construct_blueprint(settings, version):
appName=Constants.APP_NAME,
openApiSpecification=specification)
@routes.route('/measurements', methods=['GET'])
def get_all_measurements():
database = Database(settings['database']['databasePath'])
return jsonify(database.measurementAccess.get_all_measurements())
@routes.route('/measurement/<int:measurementID>', methods=['GET'])
def get_measurement(measurementID):
database = Database(settings['database']['databasePath'])
return jsonify(database.measurementAccess.get_measurement(measurementID))
@routes.route('/measurements', methods=['POST'])
@require_api_key(password=settings['api']['key'])
def addMeasurement():
try:
parameters = RequestValidator.validate(request, DeviceParameters.get_values())
database = Database(settings['database']['databasePath'])
deviceName = parameters[DeviceParameters.DEVICE.value]
if not database.deviceAccess.get_device_by_name(deviceName):
database.deviceAccess.add_device(deviceName)
device = database.deviceAccess.get_device_by_name(deviceName)
sensors = parameters[DeviceParameters.SENSORS.value]
for sensor in sensors:
sensorParams = RequestValidator.validate_parameters(sensor,
SensorParameters.get_values(),
f'sensor "{sensor}"')
sensor = __add_sensor_if_not_exists(database, int(device['id']), sensorParams)
database.measurementAccess.add_measurement(int(sensor['id']),
sensorParams[SensorParameters.VALUE.value])
except ValidationError as e:
return e.response, 400
return jsonify({'success': True})
def __add_sensor_if_not_exists(database: Database, deviceID: int, sensorParams: Dict) -> Dict[str, str]:
sensorName = sensorParams[SensorParameters.NAME.value]
sensorType = sensorParams[SensorParameters.TYPE.value]
sensor = database.sensorAccess.get_sensor_by_name_and_device_id(deviceID, sensorName)
if sensor:
return sensor
database.sensorAccess.add_sensor(deviceID, sensorName, sensorType)
return database.sensorAccess.get_sensor_by_name_and_device_id(deviceID, sensorName)
return routes
from enum import Enum
from flask import Blueprint, jsonify
from logic.Database import Database
class SensorParameters(Enum):
NAME = 'name'
TYPE = 'type'
VALUE = 'value'
@staticmethod
def get_values():
return [m.value for m in SensorParameters]
def construct_blueprint(settings):
sensors = Blueprint('sensors', __name__)
......
from enum import Enum
class DeviceParameters(Enum):
DEVICE = 'device'
SENSORS = 'sensors'
@staticmethod
def get_values():
return [m.value for m in DeviceParameters]
class SensorParameters(Enum):
NAME = 'name'
TYPE = 'type'
VALUE = 'value'
@staticmethod
def get_values():
return [m.value for m in SensorParameters]
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment