最近公司需要做数据的定时处理。使用Java的spring boot的scheduling很方便，但是搞算法的专家一般只会python。那么如何利用python定时对于只能内网访问的数据库处理，就是一个问题咯。
代码库：https://gitlab.com/goddy-basic/flask-schedule
1、读前建议资料
- python3
- docker
- pandas
- gitlab-ci
- flask
- pymysql
2、参考资料（必读）
- (cnblogs - 上官飞鸿的博客)[https://www.cnblogs.com/jackadam/p/8146624.html]
- (cnblogs - 23云恋49枫的博客)[https://www.cnblogs.com/luxiaojun/p/6567132.html]
感谢以上作者的分享🙏give me five
3、个性化
看完大佬们的博客分享，那么开始我们的个性化配置。
1. 安装依赖包
#可写进项目根目录的requirements.txt
pandas
flask
flask_apscheduler
datetime
pymysql
2.代码架构
#数据库处理层
1.DAO
#示例项目
2.task1
#工具包
3.util
#配置文件
4.config.py
#启动文件 + 接口路由
5.app.py
#定时任务启动方法汇总
6.schedule_job.py
3.数据库连接配置（pymysql）
首先，先写一个获取数据库连接池的方法，注意port一定为数值型。
def conn_mysql():
'''
获取数据库连接池
:return: mysql配置
'''
config = mysql_config()
return pymysql.connect(
host="localhost",
user="root",
password="pass",
db="mydb",
port=3306,
charset="utf8"
)
然后编写通用的数据库读写方法：
def read(sql):
'''
读操作
:param sql: sql语句
:return: 读取后数据
'''
print("【mysql】'{}'".format(sql))
return pd.read_sql(sql, conn_mysql())
def save(sql):
'''
添加/更新操作
:param sql:
:return:
'''
print("【mysql】'{}".format(sql))
conn = conn_mysql()
conn.cursor().execute(sql)
conn.commit()
4. DAO层使用
因为最近再学java，所以满脑子java的分层思想。DAO即data access object，数据访问对象是专门面向对象的数据库接口，跟数据库打交道。举个例子就是：
def test_insert(time):
sql = "INSERT INTO test (time) VALUES ('{}')".format(
time
)
mysqlUtil.save(sql)
5.业务层使用数据
然后在我们相应服务的包里调用DAO层接口
#task1.task1_insert.py
import datetime
from DAO import provider_dao
def insert():
provider_dao.test_insert(datetime.datetime.now())
是不是超像java😝
6.统一将定时任务放到schedule_job.py
因为一个定时任务可以需要调用多个业务层的方法，所以把定时任务的归口方法写在这里。
文件名字schedule_job.py 我取的,可以自己修改
def insert_test():
task1_insert.insert()
print('inserted!')
7.接口调用业务层
然后在falsk的启动py文件中调用刚写的业务层insert方法
from flask import Flask
@app.route("/insert")
def insert():
schedule_job.insert_test()
return "success"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=False)
8. 配置config.py
定时任务的配置项此文章末尾附录获取
#添加app.py可以修改此文件环境的方法
def set_env(environment):
'''
配置环境变量
:param environment:
:return:
'''
global env
env = environment
#基础类
class Config(object):
SCHEDULER_API_ENABLED = True
#生产环境
class DevConfig(Config):
JOBS = [
{
'id': 'job3',
'func': 'schedule_job:insert_test',
'args': '',
'trigger': {
'type': 'interval',
'seconds': 5
}
},
{
'id': 'job1',
'func': 'schedule_job:test_data',
'args': '',
'trigger': {
'type': 'cron',
'day_of_week': "mon-fri",
'hour': '0-23',
'minute': '0-59',
'second': '*/2'
}
}, {
'id': 'job2',
'func': 'schedule_job:test_boom',
'args': '',
'trigger': {
'type': 'interval',
'seconds': 5
}
}
]
#开发环境
class ProConfig(Config):
JOBS = [
{
'id': 'job3',
'func': 'schedule_job:insert_test',
'args': '',
'trigger': {
'type': 'interval',
'seconds': 5
}
}
]
#mysql连接，将之前的mysqlUtil中的配置型改为从此方法获取
def mysql_config():
if env == 'DevConfig':
return {
'host': 'xxx',
'user': 'xx',
'password': 'xxx',
'db': 'xxx',
'port': xxx,
'charset': 'xxx'
}
elif env == 'ProConfig':
return {
'host': 'xxx',
'user': 'xxx',
'password': 'xxx',
'db': 'xxxx',
'port': xxx,
'charset': 'xxxx'
}
else:
raise RuntimeError('no such config ...')
9.修改app.py添加schedule
中间这段是通过执行python命令时附带参数来改变开发环境和生产环境，默认是开发环境
if __name__ == '__main__':
scheduler = APScheduler()
env = sys.argv[1] if len(sys.argv) > 1 else 'DevConfig'
print("Python program begins in %s environment." % (env))
config.set_env(env)
# app.config.from_object('config.' + config.env)
app.config.from_object('config.' + env)
scheduler.init_app(app)
scheduler.start()
app.run(host='0.0.0.0', port=5000, debug=False)
10.自动切换开发环境和生产环境
首先，本地肯定会使用开发环境，那么怎么能部署的时候自动使用开发环境呢？使用Dockerfile文件来区分。
FROM python
ADD . /code
WORKDIR /code
#安装依赖
RUN pip install -r requirements.txt
#修改系统时间
RUN ln -fs /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && dpkg-reconfigure -f noninteractive tzdata
CMD ["python", "app.py", "ProConfig"]
建议看下我的代码，链接在前言。
附录
来源于https://www.cnblogs.com/luxiaojun/p/6567132.html，大佬别怪罪
网友评论