Skip to content

Commit

Permalink
Added internal token validation on endpoints.
Browse files Browse the repository at this point in the history
  • Loading branch information
@penserbjorne
penserbjorne committed May 11, 2021
1 parent 0933962 commit 3b0a6c2
Show file tree
Hide file tree
Showing 13 changed files with 1,247 additions and 937 deletions.
175 changes: 100 additions & 75 deletions app/controllers/area.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
from flask_restx import Resource, fields
import json

from app import api, isOnDev, project_dir
from app import api, isOnDev, project_dir, INTERNAL_TOKEN
from app.models.area import AreaModel as TheModel
from app.schemas.area import AreaSchema as TheSchema
from app.const import HttpStatus, EmptyValues
Expand Down Expand Up @@ -33,38 +33,48 @@
class AreaList(Resource):
@local_ns.doc('Get all the ' + CURRENT_NAME + 's')
def get(self):
try:
if isOnDev:
response = jsonify(TheModel.find_all())
response.status_code = HttpStatus.OK
else:
f = open(project_dir + CACHE_FILE, "r")
data_json = json.loads(f.read())
f.close()
response = jsonify(data_json)
response.status_code = HttpStatus.OK
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
if INTERNAL_TOKEN.compare(request.headers.get('Authorization')):
try:
if isOnDev:
response = jsonify(TheModel.find_all())
response.status_code = HttpStatus.OK
else:
f = open(project_dir + CACHE_FILE, "r")
data_json = json.loads(f.read())
f.close()
response = jsonify(data_json)
response.status_code = HttpStatus.OK
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
else:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response

@local_ns.doc('Create an ' + CURRENT_NAME)
@local_ns.expect(model_validator)
def post(self):
if not isOnDev:
if INTERNAL_TOKEN.compare(request.headers.get('Authorization')):
if not isOnDev:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_json = request.get_json()
element_data = local_schema.load(element_json)
element_data.save()
response = jsonify(element_data.json())
response.status_code = HttpStatus.CREATED
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.BAD_REQUEST
return response
else:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_json = request.get_json()
element_data = local_schema.load(element_json)
element_data.save()
response = jsonify(element_data.json())
response.status_code = HttpStatus.CREATED
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.BAD_REQUEST
return response

@local_ns.route('/<int:id>')
class Area(Resource):
Expand All @@ -73,70 +83,85 @@ class Area(Resource):
'id': 'id of the ' + CURRENT_NAME + ' to get'
})
def get(self, id):
try:
element_data = TheModel.find_by_id(id)
if element_data:
response = jsonify(element_data.json())
response.status_code = HttpStatus.OK
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
if INTERNAL_TOKEN.compare(request.headers.get('Authorization')):
try:
element_data = TheModel.find_by_id(id)
if element_data:
response = jsonify(element_data.json())
response.status_code = HttpStatus.OK
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
else:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response

@local_ns.doc('Update an ' + CURRENT_NAME + ' with the specified id',
params={
'id': 'id of the ' + CURRENT_NAME + ' to update'
})
@local_ns.expect(model_validator)
def put(self, id):
if not isOnDev:
if INTERNAL_TOKEN.compare(request.headers.get('Authorization')):
if not isOnDev:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_data = TheModel.find_by_id(id)

if element_data:
element_data.ocd_id = EmptyValues.EMPTY_STRING if request.json['ocd_id'] == EmptyValues.EMPTY_STRING else request.json['ocd_id']
element_data.name = EmptyValues.EMPTY_STRING if request.json['name'] == EmptyValues.EMPTY_STRING else request.json['name']
element_data.country = EmptyValues.EMPTY_STRING if request.json['country'] == EmptyValues.EMPTY_STRING else request.json['country']
element_data.state = EmptyValues.EMPTY_STRING if request.json['state'] == EmptyValues.EMPTY_STRING else request.json['state']
element_data.city = EmptyValues.EMPTY_STRING if request.json['city'] == EmptyValues.EMPTY_STRING else request.json['city']
element_data.district_type = EmptyValues.EMPTY_INT if request.json['district_type'] == EmptyValues.EMPTY_STRING else request.json['district_type']
element_data.parent_area_id = EmptyValues.EMPTY_INT if request.json['parent_area_id'] == EmptyValues.EMPTY_STRING else request.json['parent_area_id']
element_data.save()
response = jsonify(element_data.json())
response.status_code = HttpStatus.CREATED
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.BAD_REQUEST
return response
else:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_data = TheModel.find_by_id(id)

if element_data:
element_data.ocd_id = EmptyValues.EMPTY_STRING if request.json['ocd_id'] == EmptyValues.EMPTY_STRING else request.json['ocd_id']
element_data.name = EmptyValues.EMPTY_STRING if request.json['name'] == EmptyValues.EMPTY_STRING else request.json['name']
element_data.country = EmptyValues.EMPTY_STRING if request.json['country'] == EmptyValues.EMPTY_STRING else request.json['country']
element_data.state = EmptyValues.EMPTY_STRING if request.json['state'] == EmptyValues.EMPTY_STRING else request.json['state']
element_data.city = EmptyValues.EMPTY_STRING if request.json['city'] == EmptyValues.EMPTY_STRING else request.json['city']
element_data.district_type = EmptyValues.EMPTY_INT if request.json['district_type'] == EmptyValues.EMPTY_STRING else request.json['district_type']
element_data.parent_area_id = EmptyValues.EMPTY_INT if request.json['parent_area_id'] == EmptyValues.EMPTY_STRING else request.json['parent_area_id']
element_data.save()
response = jsonify(element_data.json())
response.status_code = HttpStatus.CREATED
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.BAD_REQUEST
return response

@local_ns.doc('Delete an ' + CURRENT_NAME + ' with the specified id',
params={
'id': 'id of the ' + CURRENT_NAME + ' to delete'
})
def delete(self, id):
if not isOnDev:
if INTERNAL_TOKEN.compare(request.headers.get('Authorization')):
if not isOnDev:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_data = TheModel.find_by_id(id)
if element_data:
element_data.delete()
response = jsonify({'message': CURRENT_NAME + ' deleted.'})
response.status_code = HttpStatus.OK
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
else:
response = jsonify({'message': 'Not allowed'})
response.status_code = HttpStatus.NOT_ALLOWED
return response
try:
element_data = TheModel.find_by_id(id)
if element_data:
element_data.delete()
response = jsonify({'message': CURRENT_NAME + ' deleted.'})
response.status_code = HttpStatus.OK
else:
response = jsonify({'message': CURRENT_NAME + ' not found.'})
response.status_code = HttpStatus.NOT_FOUND
except Exception as e:
response = jsonify({'message': e.__str__()})
response.status_code = HttpStatus.INTERNAL_ERROR
return response
return response
Loading

0 comments on commit 3b0a6c2

Please sign in to comment.