我希望使用python从单个GCP云函数提供多个路由。虽然GCP功能实际上在引擎盖下使用烧瓶，但我似乎不知道如何使用烧瓶路由系统来从单个云功能提供多条路由。
我正在做一个非常小的项目，所以我写了一个自己的快速路由器，运行良好。现在我越来越多地使用GCP函数，我要么想知道如何使用Flask路由器，要么在我的手动版本上投入更多的时间，也许是开源的，尽管当它是Flask路由的一个非常接近的副本时，它似乎是多余的，所以如果这个功能不存在，也许最好直接将它添加到Flask中。
有人有这个问题的经验吗？我猜我错过了一个隐藏在Flask中的简单函数，但如果不是，这似乎是一个相当大/常见的问题，尽管我猜GCP函数python是测试版是有原因的？
编辑：如果可能的话，我想使用Flask的手卷版本的摘要示例：
router = MyRouter()
@router.add('some/path', RouteMethod.GET)
def handle_this(req):
...
@router.add('some/other/path', RouteMethod.POST)
def handle_that(req):
...
# main entry point for the cloud function
def main(request):
return router.handle(request)
以下解决方案对我有效：
import flask
import werkzeug.datastructures
app = flask.Flask(__name__)
@app.route('some/path')
def handle_this(req):
...
@app.route('some/other/path', methods=['POST'])
def handle_that(req):
...
def main(request):
with app.app_context():
headers = werkzeug.datastructures.Headers()
for key, value in request.headers.items():
headers.add(key, value)
with app.test_request_context(method=request.method, base_url=request.base_url, path=request.path, query_string=request.query_string, headers=headers, data=request.data):
try:
rv = app.preprocess_request()
if rv is None:
rv = app.dispatch_request()
except Exception as e:
rv = app.handle_user_exception(e)
response = app.make_response(rv)
return app.process_response(response)
基于http://flask.pocoo.org/snippets/131/
得益于Guillaume Blaquiere文章的灵感和一些调整，我有了一种方法，可以使用ngrok生成一个公共URL，用于谷歌云功能的本地测试和开发。
我有两个关键文件，app.py和main.py.
我正在使用VS代码，现在可以打开应用程序了；调试当前文件"；。现在我可以在我的函数main.py中设置断点。我安装了"REST客户端"扩展，然后它使我能够配置GET和POST调用，我可以针对本地和ngrok URL运行这些调用。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#app.py
import os
from flask import Flask, request, Response
from main import callback
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def test_function():
return callback(request)
def start_ngrok():
from pyngrok import ngrok
ngrok_tunnel = ngrok.connect(5000)
print(' * Tunnel URL:', ngrok_tunnel.public_url)
if __name__ == '__main__':
if os.environ.get('WERKZEUG_RUN_MAIN') != 'true':
start_ngrok()
app.run(debug=True)
#!/usr/bin/env python3
# This file main.py can be run as a Google Cloud function and deployed with:
# gcloud functions deploy callback --runtime python38 --trigger-http --allow-unauthenticated
from flask import Response
import datetime
now = datetime.datetime.now()
def callback(request):
if request.method == 'POST': # Block is only for POST request
print(request.json)
return Response(status=200)
return Response(f'''
<!doctype html><title>Hello from webhook</title>
<body><h1>Hello! </h1><p>{now:%Y-%m-%d %H:%M}</p>
</body></html>
''', status=200)
同样适用于我的@rabelenda的简化版本：
def main(request):
with app.request_context(request.environ):
try:
rv = app.preprocess_request()
if rv is None:
rv = app.dispatch_request()
except Exception as e:
rv = app.handle_user_exception(e)
response = app.make_response(rv)
return app.process_response(response)
Martin的解决方案对我有效，直到我尝试在其中一条路由中调用request.get_json()。最终的结果是，由于数据流已经被消耗，响应被阻止在较低级别。
我在Google Cloud Run中使用functions_framework寻找解决方案时遇到了这个问题。它已经在设置一个app，您可以通过从烧瓶中导入current_app来获得它。
from flask import current_app
app = current_app
我相信functions_framework是由谷歌云功能使用的，所以它也应该在那里工作。
感谢@rabelanda上面的回答启发了我的回答，它只是调整了data/json参数，并支持InternalServerError未处理的异常处理程序：
import werkzeug.datastructures
def process_request_in_app(request, app):
# source: https://stackoverflow.com/a/55576232/1237919
with app.app_context():
headers = werkzeug.datastructures.Headers()
for key, value in request.headers.items():
headers.add(key, value)
data = None if request.is_json else (request.form or request.data or None)
with app.test_request_context(method=request.method,
base_url=request.base_url,
path=request.path,
query_string=request.query_string,
headers=headers,
data=data,
json=request.json if request.is_json else None):
try:
rv = app.preprocess_request()
if rv is None:
rv = app.dispatch_request()
except Exception as e:
try:
rv = app.handle_user_exception(e)
except Exception as e:
# Fallback to unhandled exception handler for InternalServerError.
rv = app.handle_exception(e)
response = app.make_response(rv)
return app.process_response(response)
@rabelenda的解决方案是最好的。然而，对于POST请求的阻塞问题，一点重构应该可以解决
from flask.ctx import RequestContext
def app_dispatcher(request):
with RequestContext(app, request.environ, request=request):
try:
rv = app.preprocess_request()
if rv is None:
rv = app.dispatch_request()
except Exception as e:
rv = app.handle_user_exception(e)
return app.finalize_request(rv)