cherrypy server for Flask + task scheduler and monitor
Project description
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
TaskScheduler(check_interval=5,
holidays_calendar=None,
on_job_error=None,
log_filepath=None,
log_maxsize=5*1024*1024, # 5 MB
log_backups=1)
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
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)
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())
# 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
TaskMonitor(
app,
sched,
display_name=None,
endpoint="@taskmonitor",
homepage_refresh=30,
taskpage_refresh=5)
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):
homepage_refresh (int):
taskpage_refresh (int):
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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for flask_production-2.6.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f139aa9d973bf436550e94866b70153153a59943e6239e5f106f49a8981bf13e |
|
MD5 | 8d26f5984469e103f7863320a7dbfe66 |
|
BLAKE2b-256 | 7fc935b865336158d6be997dd321e2e2047dfae072c362841c675c1921413d0b |