"""HTTP API endpoint
"""
from collections import defaultdict
import dateutil.parser
import smif
from flask import current_app, jsonify, request
from flask.views import MethodView
from smif.exception import (SmifDataError, SmifDataInputError,
SmifDataNotFoundError, SmifException,
SmifValidationError)
class SmifAPI(MethodView):
"""Implement operations for Smif
"""
def get(self, key):
"""Get smif details
version: GET /api/v1/smif/version
"""
if key == 'version':
data = smif.__version__
else:
data = {}
data['version'] = smif.__version__
response = jsonify({
'data': data,
'error': {}
})
return response
class ModelRunAPI(MethodView):
"""Implement CRUD operations for model_run configuration data
"""
def get(self, model_run_name=None, action=None):
"""Get model_runs
all: GET /api/v1/model_runs/
one: GET /api/vi/model_runs/name
"""
data_interface = current_app.config.data_interface
try:
if action is None:
if model_run_name is None:
model_runs = data_interface.read_model_runs()
if 'status' in request.args.keys():
# filtered: GET /api/v1/model_runs?status=done
data = []
for model_run in model_runs:
status = current_app.config.scheduler.get_status(model_run['name'])
if status['status'] == request.args['status']:
data.append(model_run)
else:
# all: GET /api/v1/model_runs/
data = []
data = model_runs
else:
# one: GET /api/vi/model_runs/name
data = {}
data = data_interface.read_model_run(model_run_name)
elif action == 'status':
# action: GET /api/vi/model_runs/name/status
data = {}
data = current_app.config.scheduler.get_status(model_run_name)
response = jsonify({
'data': data,
'error': {}
})
except SmifException as err:
response = jsonify({
'data': data,
'error': parse_exceptions(err)
})
return response
def post(self, model_run_name=None, action=None):
"""
Create a model_run:
- POST /api/v1/model_runs
Perform an operation on a model_run
- POST /api/v1/model_runs/<model_run_name>/<action>
Available actions are
- start: Start the model_run
- kill: Stop a model_run that is currently running
- remove: Remove a model_run that is waiting to be executed
- resume: Warm start a model_run
"""
data_interface = current_app.config.data_interface
try:
if action is None:
data = request.get_json() or request.form
data_interface.write_model_run(data)
elif action == 'start':
data = request.get_json() or request.form
args = {
'verbosity': data['args']['verbosity'],
'warm_start': data['args']['warm_start'],
'output_format': data['args']['output_format']
}
if hasattr(data_interface, 'model_base_folder'):
args['directory'] = data_interface.model_base_folder
current_app.config.scheduler.add(model_run_name, args)
elif action == 'kill':
current_app.config.scheduler.kill(model_run_name)
elif action == 'remove':
raise NotImplementedError
elif action == 'resume':
raise NotImplementedError
else:
raise SyntaxError("ModelRun action '%s' does not exist" % action)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 201
return response
def put(self, model_run_name):
"""Update a model_run:
PUT /api/v1/model_runs
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data_interface.update_model_run(model_run_name, data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 200
return response
def delete(self, model_run_name):
"""Delete a model_run:
DELETE /api/v1/model_runs
"""
data_interface = current_app.config.data_interface
data_interface.delete_model_run(model_run_name)
response = jsonify({})
return response
class SosModelAPI(MethodView):
"""Implement CRUD operations for sos_model configuration data
"""
def get(self, sos_model_name):
"""Get sos_model
all: GET /api/v1/sos_models/
one: GET /api/vi/sos_models/name
"""
# return str(current_app.config)
data_interface = current_app.config.data_interface
try:
if sos_model_name is None:
data = []
data = data_interface.read_sos_models()
else:
data = {}
data = data_interface.read_sos_model(sos_model_name)
response = jsonify({
'data': data,
'error': {}
})
except SmifException as err:
response = jsonify({
'data': data,
'error': parse_exceptions(err)
})
return response
def post(self):
"""Create a sos_model:
POST /api/v1/sos_models
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data_interface.write_sos_model(data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 201
return response
def put(self, sos_model_name):
"""Update a sos_model:
PUT /api/v1/sos_models
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data_interface.update_sos_model(sos_model_name, data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 200
return response
def delete(self, sos_model_name):
"""Delete a sos_model:
DELETE /api/v1/sos_models
"""
data_interface = current_app.config.data_interface
data_interface.delete_sos_model(sos_model_name)
response = jsonify({})
return response
class SectorModelAPI(MethodView):
"""Implement CRUD operations for sector_model configuration data
"""
def get(self, sector_model_name):
"""Get sector_models
all: GET /api/v1/sector_models/
one: GET /api/vi/sector_models/name
"""
# return str(current_app.config)
data_interface = current_app.config.data_interface
try:
if sector_model_name is None:
data = []
data = data_interface.read_models(skip_coords=True)
else:
data = {}
data = data_interface.read_model(sector_model_name, skip_coords=True)
response = jsonify({
'data': data,
'error': {}
})
except SmifException as err:
response = jsonify({
'data': data,
'error': parse_exceptions(err)
})
return response
def post(self):
"""Create a sector_model:
POST /api/v1/sector_models
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
data = check_timestamp(data)
try:
data_interface.write_model(data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 201
return response
def put(self, sector_model_name):
"""Update a sector_model:
PUT /api/v1/sector_models
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
data = check_timestamp(data)
try:
data_interface.update_model(sector_model_name, data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 200
return response
def delete(self, sector_model_name):
"""Delete a sector_model:
DELETE /api/v1/sector_models
"""
data_interface = current_app.config.data_interface
data_interface.delete_model(sector_model_name)
response = jsonify({})
return response
class ScenarioAPI(MethodView):
"""Implement CRUD operations for scenarios configuration data
"""
def get(self, scenario_name):
"""Get scenarios
all: GET /api/v1/scenarios/
one: GET /api/vi/scenarios/name
"""
# return str(current_app.config)
data_interface = current_app.config.data_interface
try:
if scenario_name is None:
data = []
data = data_interface.read_scenarios(skip_coords=True)
else:
data = {}
data = data_interface.read_scenario(scenario_name, skip_coords=True)
response = jsonify({
'data': data,
'error': {}
})
except SmifException as err:
response = jsonify({
'data': data,
'error': parse_exceptions(err)
})
return response
def post(self):
"""Create a scenario:
POST /api/v1/scenarios
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data = check_timestamp(data)
data_interface.write_scenario(data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 201
return response
def put(self, scenario_name):
"""Update a scenario:
PUT /api/v1/scenarios
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data = check_timestamp(data)
data_interface.update_scenario(scenario_name, data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 200
return response
def delete(self, scenario_name):
"""Delete a scenario:
DELETE /api/v1/scenarios
"""
data_interface = current_app.config.data_interface
data_interface.delete_scenario(scenario_name)
response = jsonify({})
return response
class DimensionAPI(MethodView):
"""Implement CRUD operations for dimensions configuration data
"""
def get(self, dimension_name):
"""Get dimensions
all: GET /api/v1/dimensions/
one: GET /api/vi/dimensions/name
"""
# return str(current_app.config)
data_interface = current_app.config.data_interface
try:
if dimension_name is None:
data = []
data = data_interface.read_dimensions(skip_coords=True)
else:
data = {}
data = data_interface.read_dimension(dimension_name, skip_coords=True)
response = jsonify({
'data': data,
'error': {}
})
except SmifException as err:
response = jsonify({
'data': data,
'error': parse_exceptions(err)
})
return response
def post(self):
"""Create a dimension:
POST /api/v1/dimensions
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data = check_timestamp(data)
data_interface.write_dimension(data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 201
return response
def put(self, dimension_name):
"""Update a dimension:
PUT /api/v1/dimensions
"""
data_interface = current_app.config.data_interface
data = request.get_json() or request.form
try:
data = check_timestamp(data)
data_interface.update_dimension(dimension_name, data)
except SmifException as err:
response = jsonify({
'message': 'failed',
'data': data,
'error': parse_exceptions(err)
})
else:
response = jsonify({"message": "success"})
response.status_code = 200
return response
def delete(self, dimension_name):
"""Delete a dimension:
DELETE /api/v1/dimensions
"""
data_interface = current_app.config.data_interface
data_interface.delete_dimension(dimension_name)
response = jsonify({})
return response
[docs]def check_timestamp(data):
"""Check for timestamp and parse to datetime object
"""
if 'stamp' in data:
try:
data['stamp'] = dateutil.parser.parse(data['stamp'])
except(ValueError):
pass
return data
[docs]def parse_exceptions(exception):
"""Parse a group of exceptions so that it can be sent over
the http-api
"""
if type(exception) == SmifDataError:
msg = defaultdict(list)
for ex in exception.args[0]:
msg[str(type(ex).__name__)].append(_parse_exception(ex))
else:
msg = {}
msg[str(type(exception).__name__)] = [_parse_exception(exception)]
return msg
def _parse_exception(ex):
"""Parse a single exception so that it can be sent over
the http-api
"""
if type(ex) == SmifValidationError:
msg = ex.args[0]
if type(ex) == SmifDataInputError:
msg = {
'component': ex.component,
'error': ex.error,
'message': ex.message,
}
if type(ex) == SmifDataNotFoundError:
msg = ex.args[0]
return msg