Using Flask Routing in GCP Function?

Question

I am looking to serve multiple routes from a single GCP cloud function using python. While GCP Functions actually use flask under the hood, I can't seem to figure out how to use the flask routing system to serve multiple routes from a single cloud function.

I was working on a very small project, so I wrote a quick router of my own which is working well. Now that I'm using GCP Functions more, I'd either like to figure out how to use the Flask router or invest more time on my hand rolled version and perhaps open source it, though it would seem redundant when it would be a very close copy of flask's routing, so perhaps it would be best to add it directly to Flask if this functionality doesn't exist.

Does anyone have any experience with this issue? I'm guessing I'm missing a simple function to use that's hidden in Flask somewhere but if not this seems like a pretty big/common problem, though I guess GCP Functions python is beta for a reason?

Edit: Abridged example of my hand rolled version that I'd like to use Flask for if possible:

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)

Most web frameworks route based on the URL path. Since you can only serve a single path per Google Cloud Function, what are you hoping to use to route the request instead? — Dustin Ingram
– Dustin Ingram, Commented Nov 26, 2018 at 21:05
I want to route on the URL path, which I am doing successfully with my hand rolled version. Sorry that wasn't clear. The point is that if the cloud function runs on example.com/my-function/ I want to ideally use flask to route for /my-function/a and /my-function/b but am currently using my own as I could figure out how to use Flask's routing to do it on the GCP function, since you aren't running an app but just getting a Flask.request object back. Basically I'd love a function that takes that request object and runs it through Flask's router, though I can't find it. — adjkant
– adjkant, Commented Nov 26, 2018 at 21:20
It sounds like this is a duplicate of stackoverflow.com/questions/51995682/…. I'd stick to the custom routing you have now, or migrate this to being an App Engine app (which now has a Python 3.7 environment which also uses Flask by default) — Dustin Ingram
– Dustin Ingram, Commented Nov 26, 2018 at 22:28
Makes sense given the global context of request in Flask, was just hoping there was some magic buried that I didn't find. In this case the cloud function is a huge money saver so I'll probably develop out the custom router and stick with the function. Appreciate the time! — adjkant
– adjkant, Commented Nov 26, 2018 at 22:51

rabelenda · Accepted Answer · 2019-04-08 14:55:27Z

2

Following solution is working for me:

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)

Based on http://flask.pocoo.org/snippets/131/

answered Apr 8, 2019 at 14:55

rabelenda

1361 silver badge7 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

xEc Over a year ago

I get Error: could not handle the request if I call handle_this/that.

DezMaeth Over a year ago

im getting the same error, any chance to get feedback on it? i think the problem is related to the route like mine us-central1-company.cloudfunctions.net/my-function

Martin · Accepted Answer · 2021-02-03 11:39:21Z

2

Simplified version of @rabelenda's that also works for me:

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)

answered Feb 3, 2021 at 11:39

Martin

3833 silver badges8 bronze badges

Comments

alexdlaird · Accepted Answer · 2020-11-05 12:59:52Z

Thanks to inspiration from Guillaume Blaquiere's article and some tweaking I have an approach that enables me to use ngrok to generate a public URL for local testing and development of Google Cloud Functions.

I have two key files, app.py and main.py.

I am using VS-Code, and can now open up app.py press F5, select "Debug the current file". Now I can set breakpoints in my function, main.py. I have the 'REST Client' extension installed, which then enables me to configure GET and POST calls that I can run against my local and ngrok urls.

#!/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)

Shane Perry · Accepted Answer · 2021-02-10 17:23:08Z

1

The solution by Martin worked for me until I tried calling request.get_json() in one of my routes. The end result was the response being blocked in a lower level due to the data stream already being consumed.

I came across this question looking for a solution using functions_framework in Google Cloud Run. It is already setting up an app which you can get by importing current_app from flask.

from flask import current_app
app = current_app

I believe functions_framework is used by Google Cloud Functions so it should also work there.

answered Feb 10, 2021 at 17:23

Shane Perry

1,0007 silver badges12 bronze badges

Comments

jacob · Accepted Answer · 2020-08-16 15:18:50Z

Thanks to @rabelenda's answer above for inspiring my answer, which just tweaks the data/json parameters, as well as enables support for an InternalServerError unhandled exception handler:

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)

nafiu · Accepted Answer · 2023-08-29 03:30:53Z

0

@rabelenda's solution is the best. However, with issue of blocking for POST requests a bit refactoring should resolve it.

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)

answered Aug 29, 2023 at 3:30

nafiu

691 silver badge4 bronze badges

Collectives™ on Stack Overflow

Using Flask Routing in GCP Function?

6 Answers 6

2 Comments

Comments

Comments

Comments

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

6 Answers 6

2 Comments

Comments

Comments

Comments

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related