Gfrktyl

I wrote this for a class project, the backend for this dog voting website. I noticed duplicate code among multiple functions I was writing to be deployed as a cloud function: they all were wrapped in try/except blocks that returned either 200 and JSON or 500 and a traceback. (I understand that it would be better to return an informative, structured error code, but this helped with debugging and we didn't have extensive error handling on the front-end: this was a loss I was willing to take).

The decorator gives each function a connection pool, uses jsonschema to validate input and output, responds to CORS OPTIONS from anywhere, and uses print statements for logging. This isn't great, but it was far easier to set up than the Google Cloud Function logging library, and everything printed to stdout during function execution is logged to GCP Stackdriver.

Here is the decorator itself:

import functools

import json

import traceback



import jsonschema

from util.get_pool import get_pool



pg_pool = None





def cloudfunction(in_schema=None, out_schema=None):

    """



    :param in_schema: the schema for the input, or a falsy value if there is no input

    :param out_schema: the schema for the output, or a falsy value if there is no output

    :return: the cloudfunction wrapped function

    """

    # Both schemas must be valid according to jsonschema draft 7, if they are provided.

    if in_schema:

        jsonschema.Draft7Validator.check_schema(in_schema)

    if out_schema:

        jsonschema.Draft7Validator.check_schema(out_schema)



    def cloudfunction_decorator(f):

        """ Wraps a function with two arguments, the first of which is a json object that it expects to be sent with the

        request, and the second is a postgresql pool. It modifies it by:

         - setting CORS headers and responding to OPTIONS requests with `Allow-Origin *`

         - passing a connection from a global postgres connection pool

         - adding logging, of all inputs as well as error tracebacks.



        :param f: A function that takes a `request` and a `pgpool` and returns a json-serializable object

        :return: a function that accepts one argument, a Flask request, and calls f with the modifications listed

        """



        @functools.wraps(f)

        def wrapped(request):

            global pg_pool



            if request.method == 'OPTIONS':

                return cors_options()



            # If it's not a CORS OPTIONS request, still include the base header.

            headers = {'Access-Control-Allow-Origin': '*'}



            if not pg_pool:

                pg_pool = get_pool()



            try:

                conn = pg_pool.getconn()



                if in_schema:

                    request_json = request.get_json()

                    print(repr({"request_json": request_json}))

                    jsonschema.validate(request_json, in_schema)

                    function_output = f(request_json, conn)

                else:

                    function_output = f(conn)



                if out_schema:

                    jsonschema.validate(function_output, out_schema)



                conn.commit()

                print(repr({"response_json": function_output}))



                response_json = json.dumps(function_output)

                # TODO allow functions to specify return codes in non-exceptional cases

                return (response_json, 200, headers)

            except:

                print("Error: Exception traceback: " + repr(traceback.format_exc()))

                return (traceback.format_exc(), 500, headers)

            finally:

                # Make sure to put the connection back in the pool, even if there has been an exception

                try:

                    pg_pool.putconn(conn)

                except NameError:  # conn might not be defined, depending on where the error happens above

                    pass



        return wrapped



    return cloudfunction_decorator





# If given an OPTIONS request, tell the requester that we allow all CORS requests (pre-flight stage)

def cors_options():

    # Allows GET and POST requests from any origin with the Content-Type

    # header and caches preflight response for an 3600s

    headers = {

        'Access-Control-Allow-Origin': '*',

        'Access-Control-Allow-Headers': 'Content-Type',

        'Access-Control-Max-Age': '3600'

    }



    return ('', 204, headers)

get_pool is here:

from os import getenv

from psycopg2 import OperationalError, connect

from psycopg2.pool import SimpleConnectionPool



INSTANCE_CONNECTION_NAME = getenv('INSTANCE_CONNECTION_NAME', "")



POSTGRES_USER = getenv('POSTGRES_USER', "")

POSTGRES_PASSWORD = getenv('POSTGRES_PASSWORD', "")

POSTGRES_NAME = getenv('POSTGRES_DATABASE', "postgres")



pg_config = {

    'user': POSTGRES_USER,

    'password': POSTGRES_PASSWORD,

    'dbname': POSTGRES_NAME

}





def get_pool():

    try:

        return __connect(f'/cloudsql/{INSTANCE_CONNECTION_NAME}')

    except OperationalError:

        # If production settings fail, use local development ones

        return __connect('localhost')





def __connect(host):

    """

    Helper functions to connect to Postgres

    """

    pg_config['host'] = host

    return SimpleConnectionPool(1, 1, **pg_config)





def get_connection(host="localhost"):

    pg_config["host"] = host

    return connect(**pg_config)

And an example of usage:

@cloudfunction(

    in_schema={"type": "string"},

    out_schema={

        "anyOf": [{

            "type": "object",

            "properties": {

                "dog1": {"type": "integer"},

                "dog2": {"type": "integer"}

            },

            "additionalProperties": False,

            "minProperties": 2,

        }, {

            "type": "null"

        }]

    })

def get_dog_pair(request_json, conn):

    [function body elided]