<s> import os <EOL> import StringIO <EOL> import traceback <EOL> from datetime import datetime <EOL> from flask import jsonify , request <EOL> from manager_rest import config <EOL> from manager_rest import utils <EOL> from manager_rest import models <EOL> from manager_rest . blueprints_manager import get_blueprints_manager <EOL> from manager_rest . constants import ( MAINTENANCE_MODE_ACTIVATED , <EOL> MAINTENANCE_MODE_STATUS_FILE , <EOL> MAINTENANCE_MODE_ACTIVATING , <EOL> MAINTENANCE_MODE_ACTIVE_ERROR_CODE , <EOL> MAINTENANCE_MODE_ACTIVATING_ERROR_CODE ) <EOL> FORBIDDEN_METHODS = [ '<STR_LIT:POST>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ALLOWED_ENDPOINTS = [ '<STR_LIT>' , <EOL> '<STR_LIT:status>' , <EOL> '<STR_LIT:version>' ] <EOL> def get_maintenance_file_path ( ) : <EOL> return os . path . join ( <EOL> config . instance ( ) . maintenance_folder , <EOL> MAINTENANCE_MODE_STATUS_FILE ) <EOL> def prepare_maintenance_dict ( status , <EOL> activated_at = '<STR_LIT>' , <EOL> remaining_executions = None , <EOL> requested_by = '<STR_LIT>' , <EOL> activation_requested_at = '<STR_LIT>' ) : <EOL> state = { '<STR_LIT:status>' : status , <EOL> '<STR_LIT>' : activated_at , <EOL> '<STR_LIT>' : remaining_executions , <EOL> '<STR_LIT>' : requested_by , <EOL> '<STR_LIT>' : activation_requested_at } <EOL> return state <EOL> def maintenance_mode_handler ( ) : <EOL> if _is_internal_request ( ) and _is_bypass_maintenance_mode ( ) : <EOL> return <EOL> index = request . endpoint . find ( '<STR_LIT:/>' ) <EOL> request_endpoint = request . endpoint [ index + <NUM_LIT:1> : ] <EOL> maintenance_file = os . path . join ( <EOL> config . instance ( ) . maintenance_folder , <EOL> MAINTENANCE_MODE_STATUS_FILE ) <EOL> if os . path . isfile ( maintenance_file ) : <EOL> state = utils . read_json_file ( maintenance_file ) <EOL> if state [ '<STR_LIT:status>' ] == MAINTENANCE_MODE_ACTIVATING : <EOL> running_executions = get_running_executions ( ) <EOL> if not running_executions : <EOL> now = str ( datetime . now ( ) ) <EOL> state = prepare_maintenance_dict ( <EOL> MAINTENANCE_MODE_ACTIVATED , <EOL> activated_at = now , <EOL> remaining_executions = None , <EOL> requested_by = state [ '<STR_LIT>' ] , <EOL> activation_requested_at = state [ <EOL> '<STR_LIT>' ] ) <EOL> utils . write_dict_to_json_file ( maintenance_file , state ) <EOL> else : <EOL> return _handle_activating_mode ( <EOL> state = state , <EOL> request_endpoint = request_endpoint ) <EOL> if _check_allowed_endpoint ( request_endpoint ) : <EOL> return <EOL> if state [ '<STR_LIT:status>' ] == MAINTENANCE_MODE_ACTIVATED : <EOL> return _maintenance_mode_error ( ) <EOL> def _handle_activating_mode ( state , request_endpoint ) : <EOL> status = state [ '<STR_LIT:status>' ] <EOL> if request_endpoint == '<STR_LIT>' : <EOL> if request . method in FORBIDDEN_METHODS : <EOL> return _return_maintenance_error ( status ) <EOL> if request_endpoint == '<STR_LIT>' '<STR_LIT>' : <EOL> return _return_maintenance_error ( status ) <EOL> if request_endpoint == '<STR_LIT>' : <EOL> if request . method in FORBIDDEN_METHODS : <EOL> return _return_maintenance_error ( status ) <EOL> if request_endpoint == '<STR_LIT>' : <EOL> if request . method in FORBIDDEN_METHODS : <EOL> return _return_maintenance_error ( status ) <EOL> if request_endpoint == '<STR_LIT>' : <EOL> if request . method in FORBIDDEN_METHODS : <EOL> return _return_maintenance_error ( status ) <EOL> def _return_maintenance_error ( status ) : <EOL> if status == MAINTENANCE_MODE_ACTIVATED : <EOL> return _maintenance_mode_error ( ) <EOL> return _activating_maintenance_mode_error ( ) <EOL> def _check_allowed_endpoint ( request_endpoint ) : <EOL> for endpoint in ALLOWED_ENDPOINTS : <EOL> if request_endpoint . startswith ( endpoint ) : <EOL> return True <EOL> return False <EOL> def get_running_executions ( ) : <EOL> executions = get_blueprints_manager ( ) . executions_list ( <EOL> is_include_system_workflows = True ) . items <EOL> running_executions = [ ] <EOL> for execution in executions : <EOL> if execution . status not in models . Execution . END_STATES : <EOL> running_executions . append ( { <EOL> '<STR_LIT:id>' : execution . id , <EOL> '<STR_LIT:status>' : execution . status , <EOL> '<STR_LIT>' : execution . deployment_id , <EOL> '<STR_LIT>' : execution . workflow_id <EOL> } ) <EOL> return running_executions <EOL> def _is_internal_request ( ) : <EOL> remote_addr = _get_remote_addr ( ) <EOL> http_host = _get_host ( ) <EOL> return all ( [ remote_addr , http_host , remote_addr == http_host ] ) <EOL> def _is_bypass_maintenance_mode ( ) : <EOL> bypass_maintenance_header = '<STR_LIT>' <EOL> return request . headers . get ( bypass_maintenance_header ) <EOL> def _get_remote_addr ( ) : <EOL> return request . remote_addr <EOL> def _get_host ( ) : <EOL> return request . host <EOL> def _maintenance_mode_error ( ) : <EOL> s_traceback = StringIO . StringIO ( ) <EOL> traceback . print_exc ( file = s_traceback ) <EOL> response = jsonify ( <EOL> { "<STR_LIT:message>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : MAINTENANCE_MODE_ACTIVE_ERROR_CODE , <EOL> "<STR_LIT>" : s_traceback . getvalue ( ) } ) <EOL> response . status_code = <NUM_LIT> <EOL> return response <EOL> def _activating_maintenance_mode_error ( ) : <EOL> s_traceback = StringIO . StringIO ( ) <EOL> traceback . print_exc ( file = s_traceback ) <EOL> response = jsonify ( <EOL> { "<STR_LIT:message>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : MAINTENANCE_MODE_ACTIVATING_ERROR_CODE , <EOL> "<STR_LIT>" : s_traceback . getvalue ( ) } ) <EOL> response . status_code = <NUM_LIT> <EOL> return response </s>
