flask-production 3.1.10

cherrypy server for Flask + task scheduler and monitor

Cherrypy prod server for Flask + parallel task scheduler

Installation

pip install flask_production

Usage example

CherryFlask

Cherrypy server on top of Flask app

CherryFlask(app, scheduler=None, silent=False)

Parameters:

• app (Flask): Flask application

• scheduler (TaskScheduler): task scheduler to run in parallel with Flask app

• silent (bool): don’t print logs

  • default False

from flask import Flask
from flask_production import CherryFlask

app = Flask(__name__)
...

cherry = CherryFlask(app)
cherry.run(host="0.0.0.0", port=8080, threads=5, debug=False)

TaskScheduler

Main class to setup, run and manage jobs

TaskScheduler(check_interval=5,
   holidays_calendar=None,
   tzname=None,
   on_job_error=None,
   log_filepath=None,
   log_maxsize=5*1024*1024, # 5 MB
   log_backups=1,
   startup_grace_mins=0, # minutes
   persist_states=True,
   state_handler=None)

Parameters:

• check_interval (int): how often to check for pending jobs

  • default 5 seconds

• holidays_calendar (holidays.HolidayBase): calendar to use for intervals like businessday

  • default US holidays

• tzname (str): name of timezone as supported by dateutil.tz

• on_job_error (func(e)): function to call if any job fails

• log_filepath (path): file to write logs to

• log_maxsize (int): byte limit per log file

  • default 5 mb (only effective if log_filepath is provided)

• log_backups (int): number of backups of logs to retain

  • default 1 (only effective if log_filepath is provided)

• startup_grace_mins (int): grace period for tasks in case a schedule was missed because of app restart

  • default 0

• persist_states (bool): store job logs and read back on app restart

  • default True (logs will be stored)

• state_handler (.state.BaseStateHandler): different handler backends to store job logs

  • default .state.FileSystemState (logs will be stored in a unique data directory)

Standalone usage

from flask_production import TaskScheduler

sched = TaskScheduler(check_interval=2)

# Run every minute
sched.every(60).do(foo)

# Run on end of every month (with strict_date False)
sched.every("31st").strict_date(False).at("08:00").do(foo)

# Run every weekday
sched.every("weekday").at("08:00").do(lambda:bar())
sched.every("weekday").at("08:00").timezone("Europe/London").do(lambda:bar())

# catch() will run on job error
example_job = sched.every("weekday").at("09:00").do(lambda:failing()).catch(lambda e: print(e))

# access job information and status as dict
print(example_job.to_dict())
print(sched.jobs[-1].to_dict()) # same job

sched.start() # starts the task scheduler and blocks

Instead of sched.start(), TaskScheduler can be run in parallel with a Flask application using CherryFlask

from flask import Flask
from flask_production import TaskScheduler, CherryFlask

app = Flask(__name__)
...

sched = TaskScheduler()
...

cherry = CherryFlask(app, scheduler=sched)
cherry.run(host="0.0.0.0", port=8080, threads=5, debug=False)

TaskMonitor

The TaskScheduler exposes a list of Job objects through the .jobs attribute

Job information and logs from the last execution are available using the .to_dict() method

TaskMonitor uses these features to provide a web interface to view and rerun tasks

TaskMonitor(
   app,
   sched,
   display_name=None,
   endpoint="@taskmonitor",
   homepage_refresh=30,
   taskpage_refresh=5,
   can_rerun=True,
   can_disable=True)

Parameters:

• app (int): Flask application

• sched (TaskScheduler): task scheduler with task definitions

• display_name (str): name of the application to be displayed

  • default app.name

• endpoint (str): URL endpoint where the taskmonitor can be viewed

  • default “@taskmonitor”

• homepage_refresh (int): home page auto refresh interval (in seconds)

  • default 30

• taskpage_refresh (int): task page auto refresh interval (in seconds)

  • default 5

• can_rerun (bool): if True adds rerun button to job page

  • default True

• can_disable (bool): if True adds disable button to job page

  • default True

from flask import Flask
from flask_production import CherryFlask, TaskScheduler
from flask_production.plugins import TaskMonitor

app = Flask(__name__)
sched = TaskScheduler(check_interval=2)

monitor = TaskMonitor(app, sched)
print(monitor._endpoint) # /@taskmonitor

# Run every minute
sched.every(60).do(foo)

cherry = CherryFlask(app, scheduler=sched)
cherry.run(host="0.0.0.0", port=8080) # localhost:8080/@taskmonitor

Example Gist here

TODO:

scheduler - function argument validation