-
-
Save gmolveau/10c80785d499c2e2d4c447343a624664 to your computer and use it in GitHub Desktop.
# please use python3 and flask > 1.0 | |
# run this with : python3 flask_separate_thread.py | |
# https://web.archive.org/web/20210329135424/https://pymotw.com/3/threading/ | |
# https://web.archive.org/web/20210329135553/https://eli.thegreenplace.net/2011/08/22/how-not-to-set-a-timeout-on-a-computation-in-python | |
import time | |
import threading | |
from flask import Flask | |
app = Flask(__name__) | |
@app.route('/') | |
def hello_world(): | |
return 'Hello, World!' | |
def web(): | |
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=5000) | |
def usb(arg1, arg2): | |
print(f"this is how to pass arguments to a thread function - {arg1} - {arg2}") | |
while True: | |
print("waiting for USB") | |
time.sleep(2) | |
if __name__ == '__main__': | |
threading.Thread(target=web, daemon=True).start() | |
threading.Thread(target=usb, daemon=True).start() | |
while True: | |
time.sleep(1) |
How I can deploy this on production if flask docs say don't use run() in a production setting?
By using a WSGI compliant server like Gunicorn (see Flask docs)
Thanks for your reply. I discovered a way to make a threaded Flask app
Can you share it please?
How I can deploy this on production if flask docs say don't use run() in a production setting?
By using a WSGI compliant server like Gunicorn (see Flask docs)
Thanks for your reply. I discovered a way to make a threaded Flask app
Can you share it please?
Well, I separated my code into files, but I will explain each one. I work with Blueprints for my route functions.
Work with apache-XAMPP as webserver and Waitress as Python WSGI server.
First, I used a project structure that used this tutorial (but it is in Spanish). My project structure is:
project_dir/
|__ entry_point.py
|__ app/
|__ __init__.py
|__ updater.py
|__ some_module/
|__ __init__.py
|__ routes.py
|__ models.py
|__ templates
|__ some_module.html
The entry_point.py
has the instance of the Flask application that calls the generate_app()
function on app/__init__.py
:
# entry_point.py
import os
from app import generate_app
settings_module = os.getenv('APP_SETTINGS_MODULE')
application = generate_app(settings_module=settings_module)
The app/__init__.py
has the function that generates the Flask application. Any other thread should be initiated here:
# app/__init__.py
import threading
from flask import Flask
from werkzeug.middleware.proxy_fix import ProxyFix
from flask_login import LoginManager
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from werkzeug.middleware.proxy_fix import ProxyFix
...
from app.updater import update # The function that runs in another thread
login_manager = LoginManager()
db = SQLAlchemy()
migrate = Migrate()
...
def generate_app(settings_module='config.DevelopmentConfig'):
app = Flask(__name__, instance_relative_config=True)
app.config.from_object(settings_module)
# WSGI PROXY CONFIGURATION
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
# MODULES INITIALIZATION
login_manager.init_app(app)
db.init_app(app)
migrate.init_app(app, db)
...
register_blueprints(app)
register_error_handlers(app)
# THREAD INITIALIZATION
threading.Thread(target=update, args=(app,), daemon=True).start()
...
return app
In the app/updater.py
file, should be written the function that runs in the other thread, that it will be called by generate_app()
in app/__init__.py
:
# app/updater.py
def update(application):
with application.app_context():
print('ANOTHER THREAD')
...
Finally, it will be started with: flask run
command or waitress-serve --listen=127.0.0.1:5000 entry_point:application
command. The Waitress command is for production.
Note: The debug mode may duplicate the Flask app
Thanks for your reply. I discovered a way to make a threaded Flask app