# -*- encoding: utf-8 -*- # @author : vincent.benoit@benserv.fr # @brief : info routes ######################################################### # Importation de modules externes # import sys, re, os import logging as log from datetime import datetime, timezone from flask import Flask, Blueprint, request, abort, jsonify, current_app from flask_api import status from flask_jwt_extended import create_access_token from flask_jwt_extended import get_jwt from flask_jwt_extended import set_access_cookies from flask_jwt_extended import get_jwt_identity from flask_jwt_extended import jwt_required import json from werkzeug.exceptions import HTTPException ######################################################### # Class et Methods # info = Blueprint('info', __name__, url_prefix='/api/configurateur') @info.errorhandler(HTTPException) def handle_exception(e): ''' return JSON instead of HTML for HTTP errors ''' response = e.get_response() # replace the body with JSON response.data = json.dumps({ 'code': e.code, 'name': e.name, 'description': e.description, }) response.content_type = "application/json" return response @info.after_request def refresh_expiring_tokens(response): ''' Using an 'after_request' callback, we refresh any token that is within 30 minutes of expiring.''' try: exp_timestamp = get_jwt()['exp'] now = datetime.now(timezone.utc) target_timestamp = datetime.timestamp(now + current_app.config['DELTA']) if target_timestamp > exp_timestamp: current_app.logger.warning("On doit recréer un token ....") access_token = create_access_token(identity=get_jwt_identity()) # refresh token in storage place if os.path.exists(os.path.join("/tmp", current_app.config['PROJECT'])): with open(os.path.join("/tmp", current_app.config['PROJECT'], get_jwt_identity()['id']), 'w') as f: f.write(access_token) # Modifiy a Flask Response to set a cookie containing the access JWT. set_access_cookies(response, access_token) return response except (RuntimeError, KeyError): return response @info.route('/infos', methods=['GET']) @jwt_required() def get_infos(): ''' Retreive informations ''' # Access the identity of the current user with get_jwt_identity current_user = get_jwt_identity() content = {} # load data from JSON database with open(current_app.config['DB_PATH'], 'r') as f: data = json.load(f) content['manufacturer'] = data['INFOS']['manufacturer'] content['operator'] = data['INFOS']['control']['operator'] content['provider'] = data['INFOS']['control']['service_provider'] content['signal_dbm'] = data['INFOS']['control']['signal_dbm'] content['signal_qos'] = data['INFOS']['control']['signal_qos'] content['sim_inserted'] = data['INFOS']['control']['sim_inserted'] content['call_ready'] = data['INFOS']['control']['call_ready'] return jsonify(content), status.HTTP_200_OK