Python flask_cors 模块，CORS 实例源码

def run(port: int,
        trained_models: Dict[str, DemoModel],
        static_dir: str = None) -> None:
    """Run the server programatically"""
    print("Starting a flask server on port {}.".format(port))

    if port != 8000:
        logger.warning("The demo requires the API to be run on port 8000.")

    # This will be ``None`` if all the relevant environment variables are not defined or if
    # there is an exception when connecting to the database.
    demo_db = PostgresDemoDatabase.from_environment()

    app = make_app(static_dir, demo_db)
    CORS(app)

    for name, demo_model in trained_models.items():
        predictor = demo_model.predictor()
        app.predictors[name] = predictor

    app.run(port=port, host="0.0.0.0")

def _start(storage_method, host, port, testing=False):
    # TODO: This should probably be more specific
    origins = "moz-extension://*"
    if testing:
        # CORS won't be supported in non-testing mode until we fix our authentication
        logger.warning("CORS is enabled when ran in testing mode, don't store any sensitive data when running in testing mode!")
        origins = "*"
    # See: https://flask-cors.readthedocs.org/en/latest/
    CORS(app, resources={r"/api/*": {"origins": origins}})

    # Only pretty-print JSON if in testing mode (because of performance)
    app.config["JSONIFY_PRETTYPRINT_REGULAR"] = testing

    db = Datastore(storage_method, testing=testing)
    app.api = ServerAPI(db=db, testing=testing)
    app.run(debug=testing, host=host, port=port, request_handler=FlaskLogHandler, use_reloader=False)

def __init__(self):
        self.rospack = RosPack()
        self.port = rospy.get_param('ui_port', 80 if os.getuid() == 0 else 5000)
        self.web_app_root = join(self.rospack.get_path('apex_playground'), 'webapp', 'static')
        self.app = Flask(__name__, static_url_path='', static_folder=self.web_app_root)
        self.cors = CORS(self.app, resources={r'/api/*': {'origins': '*'}})
        self.services = UserServices()
        self.window_length = 500
        self.display_point_interval = 1

        self.app.route('/')(self.root)
        self.app.route('/api/interests', methods=['GET'])(self.experiment_status)
        self.app.route('/api/focus', methods=['POST'])(self.update_focus)
        self.app.route('/api/time-travel', methods=['POST'])(self.time_travel)
        self.app.route('/api/reset', methods=['POST'])(self.reset)
        self.app.route('/api/assessment', methods=['POST'])(self.update_assessment)

def create_app(config_name):
    app = Flask(__name__)
    # CORS(app, supports_credentials=False, resources={r"/api/*": {"origins": "*"}})
    CORS(app, supports_credentials=True)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    db.init_app(app)
    login_manager.init_app(app)

    from .main import main as index_blueprint
    app.register_blueprint(index_blueprint, url_prefix='/')

    from .api_1_0 import api_1_0 as api_1_0_blueprint
    app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1.0')

    return app

def create_app(config_name):
    app = Flask(__name__)
    # CORS(app, supports_credentials=False, resources={r"/api/*": {"origins": "*"}})
    CORS(app, supports_credentials=True)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    db.init_app(app)
    login_manager.init_app(app)

    celery.conf.update(app.config)

    from .main import main as index_blueprint
    app.register_blueprint(index_blueprint, url_prefix='/')

    from .api_1_0 import api_1_0 as api_1_0_blueprint
    app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1.0')

    return app

def create():
    """Create application.
    """

    # Create application
    app = Flask('service', static_folder=None)
    app.config['DEBUG'] = True

    # CORS support
    CORS(app, supports_credentials=True)
    app.register_blueprint(search(), url_prefix='/metastore/')

    # Return application
    return app

def token():
    refresh_token = request.values.get('refresh_token')
    if refresh_token is None:
        abort(requests.codes.bad_request)

    if not OAUTH_HAVE_REFRESH_TOKEN:
        return dict(
            refresh_token='',
            access_token=refresh_token,
            expires_at=float('inf')
        )

    session = _create_session()

    try:
        resp = session.refresh_token(
            token_url=OAUTH_ACCESS_TOKEN_URL,
            client_id=OAUTH_CLIENT_ID,  # Why??? The session object already has it!
            client_secret=OAUTH_CLIENT_SECRET,
            refresh_token=refresh_token
        )
    except OAuth2Error as ex:
        return dict(error=ex.error)

    return dict(
        refresh_token=resp['refresh_token'],
        access_token=resp['access_token'],
        expires_at=resp['expires_at']
    )


### API routes ###

# Allow CORS requests to API routes.
# The "*" origin is more secure than specific origins because it blocks cookies.
# Cache the settings for a day to avoid pre-flight requests.

def create_app(environment=None):
        """
        Create an app instance
        """
        app = Flask('core')

        # Allow CORS for all domains on all routes
        CORS(app)

        # Config app for environment
        if not environment:
            environment = os.environ.get('BACKEND_ENVIRONMENT', 'Dev')

        app.config.from_object('core.api.settings.%s' % environment)

        # convert exceptions to JSON
        def make_json_error(ex):
            response = jsonify(
                message=str(ex)
            )
            response.status_code = (
                ex.code if isinstance(ex, HTTPException) else 500
            )

            return response

        for code in default_exceptions.items():
            app.error_handler_spec[None][code] = make_json_error

        from core.api.views.endpoints import api
        app.register_module(api)

        API.app = app

def create_app(env=None):
    """
    Bootstrap function to initialise the Flask app and config
    :return: Initialised Flask app
    """
    app = Flask(__name__)

    if env is None:
        env = os.getenv('TM_ENV', 'Dev')  # default to Dev if config environment var not set

    app.config.from_object(f'server.config.{env}Config')

    initialise_logger(app)
    app.logger.info(f'HOT Tasking Manager App Starting Up, Environment = {env}')

    db.init_app(app)
    migrate.init_app(app, db)

    app.logger.debug('Initialising Blueprints')
    from .web import main as main_blueprint
    from .web import swagger as swagger_blueprint
    app.register_blueprint(main_blueprint)
    app.register_blueprint(swagger_blueprint)

    init_flask_restful_routes(app)

    CORS(app)  # Enables CORS on all API routes, meaning API is callable from anywhere

    app.secret_key = app.config['SECRET_KEY']  # Required by itsdangeroud, Flask-OAuthlib for creating entropy
    oauth.init_app(app)

    return app

def validate_socket_id(request):
    try:
        if not request.form['socket-id']:
            raise Exception("cvfy [Error Code: 011] => field socket-id not found in the incoming request")
    except:
        raise Exception("cvfy [Error Code: 011] => field socket-id not found in the incoming request")

##########
## CORS ##
##########

def serve_flask_app(app, port, quiet=True, host=None, cors=True):
    if cors:
        CORS(app)
    if quiet:
        log = logging.getLogger('werkzeug')
        log.setLevel(logging.ERROR)
    if not host:
        host = '0.0.0.0'
    ssl_context = GenericProxy.get_flask_ssl_context()
    app.run(port=int(port), threaded=True, host=host, ssl_context=ssl_context)
    return app