Reputation: 7877
I have an application with many threads. One of them is flask, which is used to implement (auxiliary) API. It's used with low load and never exposed to the Internet, so build-in flask web server is perfectly fine.
My current code looks like this:
class API:
# ... all other stuff here, skipped
def run():
app = flask.Flask('API')
@app.route('/cmd1')
def cmd1():
self.cmd1()
@app.route('/cmd2')
def cmd2()
self.cmd2()
app.run()
I feel I done it wrong, because all docs says 'create flask app at module level'. But I don't want to do this - it messes up with my tests, and API is a small part of the larger application, which has own structure and conventions (each 'application' is a separate class running in one or more threads).
How can I use Flask inside class?
Upvotes: 62
Views: 107278
Reputation: 2335
To run in a massive class object, or to avoid threading issues with flask, here its a slightly modified version of @kostas without disturbing the main class: (this way self.initialize_web_server
can run in Multiprocess environment combined with any number of indefinite loops concurrently).
flaskserver.py
from flask import Flask
class FlaskAppWrapper(object):
app = None
def __init__(self, name):
self.app = Flask(name)
#self.endpoinAction = endpointAction
def run(self, args):
self.app.run(**args)
def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None):
self.app.add_url_rule(endpoint, endpoint_name, handler)
The main class file: Targetclass.py
import itertools
import time
from flaskserver import FlaskAppWrapper
from flask import Response, jsonify, redirect, request, url_for
class TargetClass(object):
app = None
def __init__(self, name, server_config, run_server=True):
self.server_config = server_config
if run_server:
self.initialize_web_server()
def initialize_web_server(self):
self.app = FlaskAppWrapper(__name__, self.serve)
self.app.add_endpoint(endpoint='/', endpoint_name='home', handler=self.home_page)
self.app.add_endpoint(endpoint='/action', endpoint_name='metrics', handler=self.serve_metrics)
self.app.run(self.server_config) # server_config tuple => (host, port, debug)
def home_page(self):
if request.headers.get('accept') == 'text/event-stream':
def events():
for i, c in enumerate(itertools.cycle('\|/-')):
yield "data: %s %d\n\n" % (c, i)
time.sleep(.1) # an artificial delay
return Response(events(), content_type='text/event-stream')
return redirect(url_for('static', filename='index.html'))
def serve_metrics(self):
return jsonify(self.metrics)
server_config = { 'host' : 'localhost', 'port': 8626, 'debug': True }
TargetClass(name=__name__, server_config=server_config, run_server=True)
and a static html, static/index.html
<!doctype html>
<title>MyApp v1.0.1</title>
<style>
#data {
text-align: center;
}
</style>
<script src="http://code.jquery.com/jquery-latest.js"></script>
<script>
if (!!window.EventSource) {
var source = new EventSource('/');
source.onmessage = function(e) {
$("#data").text(e.data);
}
}
</script>
<div id="data">nothing received yet</div>
Upvotes: 0
Reputation: 158
A sidenote/addition to @Kostas Pelelis Answer (Sorry can't comment yet):
For all of you who wonder how to integrate the methods of the endpoint route: have a look at the function description for app.add_url_rule.
As stated there you can use the "methods" parameter to change the default "GET" method.
Kostas Pelelis code changed to a "POST" type method would look like this:
(Example with methods integrated + Endpoint-class that returns whatever your action-function returns [a html for example]
from flask import Flask, Response, render_template
class EndpointAction(object):
def __init__(self, action):
self.action = action
self.response = Response(status=200, headers={})
def __call__(self, *args):
response = self.action()
if response != None:
return response
else
return self.response
class FlaskAppWrapper(object):
app = None
def __init__(self, name):
self.app = Flask(name)
def run(self):
self.app.run()
def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None, t_methods=None):
self.app.add_url_rule(endpoint, endpoint_name, EndpointAction(handler), methods=t_methods)
def action():
# Execute anything
print('i did something')
def returning_action():
# Returning for example an index hello world page
return render_template('index.html')
a = FlaskAppWrapper('wrap')
a.add_endpoint(endpoint='/ad', endpoint_name='ad', handler=action, req_methods=['POST'])
#just a little addition for handling of a returning actionhandler method
#-> i added another endpoint but for a returning method
a.add_endpoint(endpoint='/', endpoint_name='index_page', handler=returning_action, req_methods=['GET']
a.run()
While the templates/index.html could look like this (note render_templates expects a templates-folder in the same location as your py-file with specified htmls in it):
<html lang="en"> <head> <meta charset="UTF-8"> <title>Index Page</title> </head> <body> <h1>Hello World!</h1> </body> </html>
This index page addition is called when the index route 'ip-address-of-the-webapp/' is visited (via usual browser visit -> GET request).
*Edit: to show how it would look like if your action-methods had params (for example from a route param) here an updated version of the endpoint class and the action class
class EndpointAction(object):
def __init__(self, action):
self.action = action
self.response = Response(status=200, headers={})
def __call__(self, *args, **kwargs):
response = self.action(**kwargs)
if response != None:
return response
else
return self.response
def param_action(param):
# Execute something (print param)
print(f'i did {param}')
[...]
a.add_endpoint(endpoint='/<param>', endpoint_name='parametric_action', handler=param_action, req_methods=['GET']
[...]
Upvotes: 1
Reputation: 668
So I just came across the library Flask-Classful
which was really simple comparatively
To create a simple web app inside a class is this:
from flask import Flask
from flask_classful import FlaskView
app = Flask(__name__)
class TestView(FlaskView):
def index(self):
# http://localhost:5000/
return "<h1>This is my indexpage</h1>"
TestView.register(app,route_base = '/')
if __name__ == '__main__':
app.run(debug=True)
Handling multiple route and dynamic route is also simple
class TestView(FlaskView):
def index(self):
# http://localhost:5000/
return "<h1>This is my indexpage</h1>"
def secondpage(self):
# http://localhost:5000/secondpage
return "<h1>This is my second</h1>"
def thirdpage(self,name):
# dynamic route
# http://localhost:5000/thirdpage/sometext
return "<h1>This is my third page <br> welcome"+name+"</h1>"
TestView.register(app,route_base = '/')
Adding own route name with a different method that is also possible
from flask_classful import FlaskView,route
class TestView(FlaskView):
def index(self):
# http://localhost:5000/
return "<h1>This is my indexpage</h1>"
@route('/diffrentname')
def bsicname(self):
# customized route
# http://localhost:5000/diffrentname
return "<h1>This is my custom route</h1>"
TestView.register(app,route_base = '/')
This gives the potential to create separate class and handlers for a separate dependent and independent process and just import them as a package to run on the main file or wrapper file
from package import Classname
Classname.register(app,route_base = '/')
which is really simple and object-oriented
Upvotes: 16
Reputation: 15769
Here is an example of mixing class and routing that seems reasonable to me. See also https://github.com/WolfgangFahl/pyFlaskBootstrap4/issues/2 (where i am a committer)
This design has been criticized so in the project there are some improvements to this code.
'''
Created on 27.07.2020
@author: wf
'''
from flask import Flask
from frontend.WikiCMS import Frontend
from flask import render_template
import os
class AppWrap:
def __init__(self, host='0.0.0.0',port=8251,debug=False):
self.debug=debug
self.port=port
self.host=host
scriptdir=os.path.dirname(os.path.abspath(__file__))
self.app = Flask(__name__,template_folder=scriptdir+'/../templates')
self.frontend=None
def wrap(self,route):
if self.frontend is None:
raise Exception("frontend is not initialized")
content,error=self.frontend.getContent(route);
return render_template('index.html',content=content,error=error)
def run(self):
self.app.run(debug=self.debug,port=self.port,host=self.host)
pass
def initFrontend(self,wikiId):
frontend=Frontend(wikiId)
frontend.open()
appWrap=AppWrap()
app=appWrap.app
@app.route('/', defaults={'path': ''})
@app.route('/<path:route>')
def wrap(route):
return appWrap.wrap(route)
if __name__ == '__main__':
appWrap.run()
Upvotes: 1
Reputation: 875
To complete Kostas Pelelis's answer, because I had some difficulty to find the why the Response wasn't directly using the Action returned value.
Here is another version of FLASK class without decorators :
class EndpointAction(object):
def __init__(self, action):
self.action = action
def __call__(self, *args):
# Perform the action
answer = self.action()
# Create the answer (bundle it in a correctly formatted HTTP answer)
self.response = flask.Response(answer, status=200, headers={})
# Send it
return self.response
class FlaskAppWrapper(object):
def add_all_endpoints(self):
# Add root endpoint
self.add_endpoint(endpoint="/", endpoint_name="/", handler=self.action)
# Add action endpoints
self.add_endpoint(endpoint="/add_X", endpoint_name="/add_X", handler=self.add_X)
# you can add more ...
def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None):
self.app.add_url_rule(endpoint, endpoint_name, EndpointAction(handler))
# You can also add options here : "... , methods=['POST'], ... "
# ==================== ------ API Calls ------- ====================
def action(self):
# Dummy action
return "action" # String that will be returned and display on the webpage
# Test it with curl 127.0.0.1:5000
def add_X(self):
# Dummy action
return "add_X"
# Test it with curl 127.0.0.1:5000/add_X
Upvotes: 10
Reputation: 1342
Although this works it doesn't feel compliant with the Flask style guide. If you need to wrap a Flask application inside your project, create a separate class to your needs and add functions that should be executed
from flask import Flask, Response
class EndpointAction(object):
def __init__(self, action):
self.action = action
self.response = Response(status=200, headers={})
def __call__(self, *args):
self.action()
return self.response
class FlaskAppWrapper(object):
app = None
def __init__(self, name):
self.app = Flask(name)
def run(self):
self.app.run()
def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None):
self.app.add_url_rule(endpoint, endpoint_name, EndpointAction(handler))
def action():
# Execute anything
a = FlaskAppWrapper('wrap')
a.add_endpoint(endpoint='/ad', endpoint_name='ad', handler=action)
a.run()
Some things to note here:
EndpointAction
is supposed to be a wrapper that will execute your function and generate an empty 200 response. If you want you can edit the functionality__call__
method definedUpvotes: 54