Two obvious isues for Python Cloud Functions are
- Local Development Testing
- Avoid multiple functions in single file
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 loggingimport sysfrom functools import wrapsfrom flask import abort, jsonify, make_response, render_templatefrom app.settings import PROJECT_IDlog = logging.getLogger(__name__)IS_DEV = Falsedef 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_mappingdef 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 = Nonedef 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_dbdatastore_db = Nonedef 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_dbdef 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_templatedef 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 loggingimport sysimport appIS_DEV = Falseif __name__ == '__main__': IS_DEV = Trueapp.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: