flask快速搭建

工程结构如下，全局的static和template目录：

flask快速搭建

1、main.py,读取设置，注册蓝图即各个模块，然后初始化数据库orm，app运行

#coding=utf-8
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker,scoped_session
from flask import Flask

from apps.home import home
from config import config


app = Flask(__name__)
app.config.from_object('settings')
app.register_blueprint(home, url_prefix='/')

if __name__ == '__main__':
    engine = create_engine(config['database']['engine_url'], **config['database']['engine_setting'])
    config['database']['engine'] = engine
    db_poll = scoped_session(sessionmaker(bind=engine))
    session = db_poll
    from apps.home.model import DbBase
    DbBase.metadata.create_all(bind=engine)
    config['database']['session'] = session

    app.run(host='0.0.0.0', port=8104)

2、settings.py

DEBUG=True
SECRET_KEY='sufeiabcd'
SESSION_COOKIE_NAME='session'
PREFERRED_URL_SCHEME='http'
JSONIFY_MIMETYPE='application/json'
TEMPLATES_AUTO_RELOAD=True

3、数据库配置

# coding=utf-8
import os
from sqlalchemy.pool import StaticPool
import logging
import json

# 数据库配置
database_config = dict(
    engine=None,
    # engine_url='postgresql+psycopg2://mhq:[email protected]:5432/blog',
    # 如果是使用mysql+mysqldb，在确认所有的库表列都是uft8编码后，依然有字符编码报错，
    # 可以尝试在该url末尾加上queryString charset=utf8
    engine_url='sqlite:////opt/wizard/e3cwizard.db',
    engine_setting=dict(
        echo=False,  # print sql
        echo_pool=False,
        # 设置7*60*60秒后回收连接池，默认-1，从不重置
        # 该参数会在每个session调用执行sql前校验当前时间与上一次连接时间间隔是否超过pool_recycle，如果超过就会重置。
        # 这里设置7小时是为了避免mysql默认会断开超过8小时未活跃过的连接，避免"MySQL server has gone away”错误
        # 如果mysql重启或断开过连接，那么依然会在第一次时报"MySQL server has gone away"，
        # 假如需要非常严格的mysql断线重连策略，可以设置心跳。
        # 心跳设置参考https://*.com/questions/18054224/python-sqlalchemy-mysql-server-has-gone-away
        pool_recycle=25200,
        connect_args={'check_same_thread': False},
        poolclass=StaticPool,
        # pool_size=20,
        # max_overflow=20,
    ),
    session=None,
)

# 站点相关配置以及tornado的相关参数
config = dict(
    debug=False,
    log_level="ERROR",
    log_console=False,
    log_file=True,
    log_file_path="/var/wizard/logs",  # 末尾自动添加 @端口号.txt_日期
    compress_response=True,
    xsrf_cookies=True,
    cookie_secret="kjsdhfweiofjhewnfiwehfneiwuhniu",
    login_url="/login",
    port=8103,
    max_threads_num=50,
    database=database_config,
    master=True,  # 是否为主从节点中的master节点, 整个集群有且仅有一个,(要提高可用性的话可以用zookeeper来选主,该项目就暂时不做了)
    navbar_styles={"inverse": "魅力黑", "default": "优雅白"},  # 导航栏样式
    default_avatar_url="identicon",
    application=None,  # 项目启动后会在这里注册整个server，以便在需要的地方调用，勿修改
    async_do = None,
    progress = 0,#安装进度
    result = '',#安装结果
    component=json.load(open("./static/resource/component.json","r")),  #模块名和安装包名，依赖包名的映射表
    packet_name=json.load(open("./static/resource/packet_name.json","r"))   #模块名和安装包名的映射表
)



settings = dict(
    template_path=os.path.join(os.path.dirname(__file__), "template"),
    static_path=os.path.join(os.path.dirname(__file__), "static"),
    compress_response=config['compress_response'],
    xsrf_cookies=config['xsrf_cookies'],
    cookie_secret=config['cookie_secret'],
    login_url=config['login_url'],
    debug=config['debug']
)

logging.basicConfig(filename='/opt/wizard/basic.log')
logger = logging.getLogger('main')
logger.setLevel(logging.DEBUG)

fh = logging.FileHandler('/opt/wizard/log.log', encoding='utf-8')
fh.setLevel(logging.DEBUG)
fh.setFormatter(logging.Formatter('[%(asctime)s -%(levelname)s -%(filename)s:%(lineno)d -%(funcName)s]:%(message)s'))
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(logging.Formatter('[%(asctime)s -%(levelname)s -%(filename)s:%(lineno)d -%(funcName)s]:%(message)s'))
logger.addHandler(fh)
logger.addHandler(ch)

4、每个模块的__init__.py

from flask import Blueprint

home = Blueprint(
    'home',
    __name__,
    template_folder='templates',
    static_folder='static'
)

from . import view

5、视图举例

from flask import Flask,request,redirect,session,render_template,make_response
from . import home

@home.route('/', methods=['GET','POST'], endpoint='index')
def index():
    if request.method == 'GET':
        return render_template('index.html'), 201, {'author':'sufei'}  #{{ url_for('static', filename='css/commons.css') }}'  {% for i in list %} {% endfor %}  {{ v }}
    else:
        response = make_response('hello world')
        response.headers['xxx']='flask'
        return response
        # return "Index"
        # return render_template('index.html'), 201, {'author':'sufei'}
        # return redirect('/')   #return(url_for('index')
        # return response, status, headers  #可缺省
        # return render_template()   可传入python函数func=test，或者用@app.template_global()，@app.template_filter() 装饰函数直接使用
        # return jsonify(name='alex',age='18')

相关推荐