"""Application Handlers. Requests are redirected to handlers, which are responsible for getting information from the URL and passing it down to the logic layer. The way each layer talks to each other is through Response objects which defines the type status of the data and the data itself. Please note: the Orchard uses the term handlers over views as convention for clarity See: oto.response for more details. """ from flask import Response, g, jsonify, request from owsrequest import flask_request from owsresponse import response from owsresponse.adaptors.flask import flaskify from store import config from store.api import app from store.constants import constants, error from store.logic import store from store.models import store as store_model from store.utils import authorization from store.utils.validations import sanitize_sort_params, validate_profile_access @app.route(config.HEALTH_CHECK, methods=["GET"]) def health() -> Response: """Check the health of the application.""" return jsonify({"status": "ok"}) @app.errorhandler(Exception) def exception_handler(exc: Exception) -> Response: """Handle error when uncaught exception is raised. Default exception handler. Note: Exception will also be sent to Sentry if config.SENTRY is set. Returns: flask.Response: A 500 response with JSON 'code' & 'message' payload. """ message = ( "The server encountered an internal error " "and was unable to complete your request." ) g.log.exception(exc) return flaskify(response.create_fatal_response(message)) @app.route("/stores", methods=["GET"]) def get_stores() -> Response: """Fetch all stores. Returns: flask.Response: on successful, 200 status with JSON body. """ access_rule_decision = flask_request.verify_rules_access_standalone(request) if not access_rule_decision: authorized = authorization.pdp_authorize_resource( # This resource type is all-or-nothing, permissions-wise, so id doesn't matter resource_id=0, resource_type=constants.STORE_PDP_RESOURCE_TYPE, ) if not authorized: return flaskify( response.create_error_response( code=error.ERROR_CODE_FORBIDDEN, message=error.ERROR_MESSAGE_FORBIDDEN_USER, status=403, ) ) orchard_identity_id = request.headers.get(constants.ORCHARD_IDENTITY_ID) jwt_identity_id = g.request_context.jwt_identity_id if not orchard_identity_id and not jwt_identity_id: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_CODE_AUTHORIZATION, 400 ) ) validate_access = validate_profile_access(request) if not validate_access and not authorization.is_jwt_identity_authorized( jwt_identity_id ): return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_DONT_HAVE_PERMISSIONS, status=401, ) ) supports_timed_release = request.args.get("supports_timed_release") classification_id = request.args.get("classification_id") status_arg = request.args.get("status", None) statuses = status_arg.split(",") if status_arg else None stores = store.get_stores( supports_timed_release={"true": True, "false": False, None: None}.get( supports_timed_release.lower() if supports_timed_release else None ), classification_id=int(classification_id) if classification_id else None, statuses=statuses, **sanitize_sort_params(request), ) return flaskify(response.Response({"total": len(stores), "items": stores})) @app.route("/stores/dataloader", methods=["POST"]) def get_stores_dataloader() -> Response: """ Fetch stores by store IDs. Returns: flask.Response: on successful, 200 status with JSON body. """ access_rule_decision = flask_request.verify_rules_access_standalone(request) if not access_rule_decision: authorized = authorization.pdp_authorize_resource( # This resource type is all-or-nothing, permissions-wise, so id doesn't matter resource_id=0, resource_type=constants.STORE_PDP_RESOURCE_TYPE, ) if not authorized: return flaskify( response.create_error_response( code=error.ERROR_CODE_FORBIDDEN, message=error.ERROR_MESSAGE_FORBIDDEN_USER, status=403, ) ) orchard_identity_id = request.headers.get(constants.ORCHARD_IDENTITY_ID) if not orchard_identity_id: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_CODE_AUTHORIZATION ) ) validate_access = validate_profile_access(request) if not validate_access: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_DONT_HAVE_PERMISSIONS, status=401, ) ) store_ids = request.get_json() return flaskify(response.Response(store.get_stores_by_ids(store_ids))) @app.route("/stores/distribution_types/dataloader", methods=["POST"]) def get_store_distribution_types_dataloader() -> Response: """ Fetch store distribution types by store IDs. Returns: flask.Response: on successful, 200 status with JSON body. """ access_rule_decision = flask_request.verify_rules_access_standalone(request) if not access_rule_decision: return flaskify( response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message="Unauthorized", status=401 ) ) orchard_identity_id = g.request_context.identity_id if not orchard_identity_id: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_CODE_AUTHORIZATION ) ) validate_access = validate_profile_access(request) if not validate_access: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_DONT_HAVE_PERMISSIONS, status=401, ) ) store_ids = request.get_json() return flaskify( response.Response( store_model.get_store_distribution_types_by_store_ids(store_ids) ) ) @app.route("/stores/classifications/dataloader", methods=["POST"]) def get_store_classifications_dataloader() -> Response: """ Fetch store classifications by store IDs. Returns: flask.Response: on successful, 200 status with JSON body. """ access_rule_decision = flask_request.verify_rules_access_standalone(request) if not access_rule_decision: return flaskify( response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message="Unauthorized", status=401 ) ) orchard_identity_id = request.headers.get(constants.ORCHARD_IDENTITY_ID) if not orchard_identity_id: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_CODE_AUTHORIZATION ) ) validate_access = validate_profile_access(request) if not validate_access: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_DONT_HAVE_PERMISSIONS, status=401, ) ) store_ids = request.get_json() return flaskify( response.Response(store.get_store_classifications_by_store_ids(store_ids)) ) @app.route("/substores/dataloader", methods=["POST"]) def get_substores_dataloader() -> Response: """Fetch substores by store IDs. Returns: flask.Response: on successful, 200 status with JSON body. """ access_rule_decision = flask_request.verify_rules_access_standalone(request) if not access_rule_decision: return flaskify( response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message="Unauthorized", status=401 ) ) orchard_identity_id = g.request_context.identity_id if not orchard_identity_id: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_CODE_AUTHORIZATION ) ) validate_access = validate_profile_access(request) if not validate_access: return flaskify( response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_DONT_HAVE_PERMISSIONS, status=401, ) ) store_ids = request.get_json() return flaskify( response.Response(message=store_model.get_substores_by_ids(store_ids)) )