flask+schedule+开发环境&生产环境自动切换

最近公司需要做数据的定时处理。使用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，大佬别怪罪