File Structure for Google Cloud Functions (Python): Support for Development and Split Independent Files

June 5, 2019

Two obvious isues for Python Cloud Functions are

Structure

I recommend the following structure

.
├── app
│   ├── functions
│   │   ├── test_hello_world.py
│   │   └── test_logging.py
│   ├── __init__.py'
│   └── settings.py
├── keys
│   ├── PROJECT_ID-datastore.json
│   └── PROJECT_ID-firebase-adminsdk.json
├── templates
│   └── hello_world.html
├── dev.py
├── main.py
├── README.md
└── requirements.txt

Common/Shared files

app/__init__.py store common shared variables and methods used by cloud functions.

import logging
import sys
from functools import wraps
from flask import abort, jsonify, make_response, render_template
from app.settings import PROJECT_ID

log = logging.getLogger(__name__)

IS_DEV = False

def init(debug):
    global IS_DEV
    IS_DEV = debug
    if debug:
        if not log.handlers:
            log.setLevel(logging.DEBUG)
            formatter = logging.Formatter(fmt="%(asctime)s %(levelname)s %(module)s: %(message)s", datefmt="%H:%M:%S")
            handler = logging.StreamHandler(sys.stdout)
            handler.setLevel(logging.DEBUG)
            handler.setFormatter(formatter)

            log.addHandler(handler)

# https://cloud.google.com/apis/design/errors#http_mapping
def json_abort(status_code, message, details=None):
    data = {
        'error': {
            'code': status_code,
            'message': message
        }
    }
    if details:
        data['error']['details'] = details
    response = jsonify(data)
    response.status_code = status_code
    abort(response)

def html_abort(status_code, message):
    response = make_response(render_template('abort.html', message=message, TITLE='Error'), status_code)
    # response.status_code = status_code
    abort(response)

firestore_db = None

def init_firestore():
    import firebase_admin
    from firebase_admin import credentials
    from firebase_admin import firestore

    global firestore_db
    if firestore_db:
        return firestore_db

    if not IS_DEV:
        cred = credentials.ApplicationDefault()
    else:
        cred = credentials.Certificate(f"keys/{PROJECT_ID}-firebase-adminsdk.json")
    default_app = firebase_admin.initialize_app(cred, {
      'projectId': {PROJECT_ID}
    })

    firestore_db = firestore.client()
    return firestore_db

datastore_db = None

def init_datastore():
    from google.cloud import datastore

    global datastore_db
    if IS_DEV:
        datastore_db = datastore.Client.from_service_account_json(f"keys/{PROJECT_ID}-datastore.json")
    else:
        datastore_db = datastore.Client()

    return datastore_db

def firebase_auth_required(f):
    @wraps(f)
    def wrapper(request):
        authorization = request.headers.get('Authorization')
        id_token = None
        if authorization and authorization.startswith('Bearer '):
            id_token = authorization.split('Bearer ')[1]
        else:
            json_abort(401, message="Invalid authorization")

        try:
            decoded_token = auth.verify_id_token(id_token)
        except Exception as e: # ValueError or auth.AuthError
            json_abort(401, message="Invalid authorization")
        return f(request, decoded_token)
    return wrapper

NOTE: For datastore and firetore, I am connecting to production server for local development testing. I haven’t tried datastore emulator and firestore emulator.

Example of app/settings.py

PROJECT_ID = '...'
APP_NAME = '...'

SENDGRID_API_KEY = '...'

You could have other shared files like app/forms.py, app/models.py, etc.

Cloud Functions

Each cloud functions is created as individual file in app/functions directory.

Example of app/functions/test_hello_world.py.

from flask import render_template

def test_hello_world(request):
    return render_template('test_hello_world.html')

templates/test_hello_world.html

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Test Hello World</title>
  </head>
  <body>
    <h1>Test Hello World</h1>
  </body>
</html>

Deployment

Edit main.py.

from app.functions.test_hello_world import *
# from app.functions.test_logging import *

Deploy

gcloud functions deploy test_hello_world --runtime python37 --trigger-http --project PROJECT_ID

Local Development Testing

Edit dev.py.

import logging
import sys
import app

IS_DEV = False
if __name__ == '__main__':
    IS_DEV = True

app.init(debug=IS_DEV)


from app.functions.test_hello_world import *
from app.functions.test_logging import *

if IS_DEV:
    from flask import Flask, request
    app = Flask(__name__)

    '''
    @app.route('/', methods=['POST', 'GET'])
    def test():
        return test_message(request)
    '''

    functions = [
        'test_hello_world',
        'test_logging'
        ]
    # app.add_url_rule(f'/test_message', 'test_message', test_message, methods=['POST', 'GET'], defaults={'request': request})
    for function in functions:
        app.add_url_rule(f'/{function}', function, locals()[function], methods=['POST', 'GET'], defaults={'request': request})

    app.run(host='127.0.0.1', port=8088, debug=True)

Start development server

python dev.py

NOTE: Make sure you are using Python 3.7.

Testing

http://127.0.0.1:8088/test_hello_world

References:

This work is licensed under a
Creative Commons Attribution-NonCommercial 4.0 International License.