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")
python类CORS的实例源码
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
