Last active
March 17, 2023 07:55
-
-
Save leforthomas/a094acfd6534a6c7b877fd612b8cdf68 to your computer and use it in GitHub Desktop.
Dead simple way to create a dynamically extensible Flask server
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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