PyTask

一个基于数据库的定时任务组件, 在指定的时间点执行任务, 超实用!!!
不同于sched、schedule、crontab等执行周期性任务, pytask是执行定时任务。

安装

pip install https://github.com/ziXiong/pytask.git@master#egg=pytask

或

git clone https://github.com/ziXiong/pytask.git
cd pytask & python setup.py install

原理简介

pytask把任务的执行时间和相关数据存储在数据库, 在另一个线程中循环地取出到了指定时间的任务并执行。

初始化数据库(第一次使用前)

完整的例子

pytask依赖于SQLAchemy存储任务信息到数据库

import pytask

# config dict, like you config your SQLAlchemy db.
sqlchemy_db = dict(
    drivername='mysql+mysqlconnector',
    host='localhost',
    username='username',
    password='password',
    database='mydatabase',
)

pytask.config(sqlchemy_db)

# do this, then you can see a table `t_task` in your database.
pytask.init_db()

使用步骤

完整的例子

步骤一: 注册任务

例如一个30分钟后订单过期的任务

# register a task handler
class OrderTimeoutHandler(pytask.TaskHandler):
    def handle(self, task):
        data = json.loads(task.biz_ext) 
        ### do the timing task ###
        # order = get_order_by_id(data['id'])
        # order.set_timeout(True)

    def get_biz_code(self):
        return 'order_timeout'  # this code identify a task

pytask.register_handler(OrderTimeoutHandler)

biz_code标识了一个任务(业务码), 在添加任务时要填写相应的code, pytask才会找到handler.

步骤二: 添加任务

# add an task
timeout_time = datetime.now() + timedelta(minutes=30)
data = dict(id=1)  # whatever data you need when calling handler.
pytask.add_task(Task(biz_code='order_timeout', when=timeout_time, biz_ext=json.dumps(data)))

添加一条任务, biz_code为任务的标识(业务码), 对应注册任务时的biz_code, when指定任务被执行的时间, biz_ext传入任务执行时需要的数据。

步骤三: 启动pytask

# start pytask. pytask will run in another thread.
pytask.start(daemon=False)

在另一个线程中启动task的轮询, pytask将每隔10秒查一次数据库,找出那些到了执行时间的任务并调用相应的handler执行.

试试吧~