Skip to content

Instantly share code, notes, and snippets.

@leforthomas
Last active March 17, 2023 07:55
Show Gist options
  • Save leforthomas/a094acfd6534a6c7b877fd612b8cdf68 to your computer and use it in GitHub Desktop.
Save leforthomas/a094acfd6534a6c7b877fd612b8cdf68 to your computer and use it in GitHub Desktop.
Dead simple way to create a dynamically extensible Flask server
import sys
from flask import Flask, request
# initialize our Flask application
app= Flask(__name__)
# the use case is to translate and forward incoming requests to other servers and return the response
# we want the server to be hot pluggable, ie add new "adapters" on the fly
# for the below code to work you need
# an adapter folder (we use the local adapters/ folder here)
# in this folder add an __init__.py file
# in the init file add the following code
#
# #!/usr/bin/env python
# import os, pkgutil
# __all__ = list(module for _, module, _ in pkgutil.iter_modules([os.path.dirname(__file__)]))
#
# this will automatically import all the files added in the directory
# this method allows the user to upload new scripts
# it writes directly into the local folder so you'll want to change that
@app.route("/adapter/<adapterid>/upload", methods=["POST"])
def upload(adapterid):
posted_data = request.get_json()
# check adapterid is valid
try:
check = int(adapterid)
except:
raise 'Adapter id is not a valid integer'
# TODO - check script content is valid
codeContent = posted_data['code']
# update the adapter file
with open('adapters/adapter_' + adapterid, 'w') as file:
file.write(codeContent)
return {}
# below are two examples of queries and how to retrieve the adapter and call it
# this expects the adapter to implement the common functions as defined by the service
@app.route("/adapter/<adapterid>/testpost", methods=["POST"])
def testpost(adapterid):
posted_data = request.get_json()
adapter = get_adapter(adapterid)
return adapter.testpost(posted_data)
@app.route("/adapter/<adapterid>/check", methods=["GET"])
def testget(adapterid):
adapter = get_adapter(adapterid)
return adapter.testget()
# this shows how to retrieve the adapter
def get_adapter(adapterid):
return sys.modules.get('adapters.adapter_' + adapterid)
# main thread of execution to start the server
# you can remove the debug setting but make sure you have the use_reloader setting to true
if __name__=='__main__':
app.run(debug=True, use_reloader=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment