Flask项目搭建及部署(最全教程)

#全局通用配置类
class Config(object):
    """项目配置核心类"""
    #调试模式
    DEBUG=False
​
    # 配置日志
    # LOG_LEVEL = "DEBUG"
    LOG_LEVEL = "INFO"
​
​
    # 配置redis
    # 项目上线以后，这个地址就会被替换成真实IP地址，mysql也是
    REDIS_HOST = 'your host'
    REDIS_PORT = your port
    REDIS_PASSWORD = 'your password'
    REDIS_POLL = 10
​
    #数据库连接格式
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://user:password@localhost:3306/test?charset=utf8"
    # 动态追踪修改设置，如未设置只会提示警告
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # 查询时会显示原始SQL语句
    SQLALCHEMY_ECHO = False
    # 数据库连接池的大小
    SQLALCHEMY_POOL_SIZE=10
    #指定数据库连接池的超时时间
    SQLALCHEMY_POOL_TIMEOUT=10
    # 控制在连接池达到最大值后可以创建的连接数。当这些额外的 连接回收到连接池后将会被断开和抛弃。
    SQLALCHEMY_MAX_OVERFLOW=2
​
​
    #rabbitmq参数配置
    RABBITUSER="user"
    RABBITPASSWORD="password"
    RABBITHOST="your ip"
    RABBITPORT=your port

from . import Config
​
class DevelopmentConfig(Config):
    '开发模式下的配置'
    # 查询时会显示原始SQL语句
    SQLALCHEMY_ECHO = True

from . import Config
​
class ProductionConfig(Config):
    """生产模式下的配置"""
    DEBUG = False

# 主应用的根目录
app = Flask(__name__)
​
config = {
    'dev': DevelopmentConfig,
    'prop': ProductionConfig,
}
​
# 设置配置类
Config = config['dev']
​
# 加载配置
app.config.from_object(Config)
​
# 创建数据库连接对象
db = SQLAlchemy(app)

# 配置对象，里面定义需要给 APP 添加的一系列配置
class Config(object):
    DEBUG = True
​
​
app = Flask(__name__)
​
# 从配置对象中加载配置
app.config.from_object(Config)
app.run()

app = Flask(__name__)
​
# 从配置对象中加载配置
app.config.from_pyfile("config.ini")
app.run()

app = Flask(__name__)
# 从环境变量中加载
app.config.from_envvar("FLASKCONFIG")
app.run()

from application import db
​
class Wdtest(db.Model):
    __tablename__ = "wdtest" #设置表名
    id = db.Column(db.String(100), primary_key=True, comment="主键ID")
    name = db.Column(db.String(20), index=True, comment="姓名" )
    age = db.Column(db.Integer, default=True, comment="年龄")

#给app取别名为 'index'
index_blu=Blueprint('index',__name__,template_folder='templates',static_folder='static')
​
from .views import *

from application.settings.dev import DevelopmentConfig
from application.settings.prop import ProductionConfig
​
# 主应用的根目录
app = Flask(__name__)
​
config = {
    'dev': DevelopmentConfig,
    'prop': ProductionConfig,
}
​
# 设置配置类
Config = config['dev']
​
# 加载配置
app.config.from_object(Config)
​
# 创建数据库连接对象
db = SQLAlchemy(app)
​
# todo 注册蓝图
from .apps.app1 import index_blu
app.register_blueprint(index_blu, url_prefix='/index')

def __init__(
        self,
        import_name,
        static_url_path=None,
        static_folder="static",
        static_host=None,
        host_matching=False,
        subdomain_matching=False,
        template_folder="templates",
        instance_path=None,
        instance_relative_config=False,
        root_path=None,
    ):pass

student = Wdtest(id=ids , name=name, age=age)
try:
    application.db.session.add(student)
    application.db.session.commit()
except:
    # 事務回滾
    application.db.session.rollback()

    user = Wdtest.query.first()
    application.db.session.delete(user)
    application.db.session.commit()

user = Wdtest.query.first()
user.name = name
try:
    application.db.session.commit()
except:
    # 事務回滾
    application.db.session.rollback()

# 查询所有⽤户数据
user_list=Wdtest.query.all()
​
# 查询有多少个⽤户
user_list_num=Wdtest.query.count()
# 查询第1个⽤户
user=Wdtest.query.first()
# 查询id为3的⽤户[3种⽅式]
user=Wdtest.query.get(3)  # 根据主键查询
user_list=Wdtest.query.filter_by(id=3).all()  # 以关键字实参形式进行匹配字段
user_list=Wdtest.query.filter(Wdtest.id == 3).all()  # 以恒等式形式匹配字段
​
# 查询名字结尾字符为g的所有⽤户
Wdtest.query.filter(Wdtest.name.endswith('g')).all()
​
# 查询名字包含‘wa'的所有项目
user_list=Wdtest.query.filter(Wdtest.name.contains('wa')).all()
# 模糊查询
user_list =Wdtest.query.filter(Wdtest.name.like('%a%')).all()
# 查询名字wa开头和age为20的所有⽤户[2种⽅式]
user_list=Wdtest.query.filter(Wdtest.name.startswith('wa'),Wdtest.age == 20).all()
user_list=Wdtest.query.filter(and_(Wdtest.name.startswith('wa'), Wdtest.age == 20)).all()
​
# 非条件查询查询名字不等于wade的所有⽤户[2种⽅式]
user_list=Wdtest.query.filter(not_(Wdtest.name == 'wade')).all()
user_list=Wdtest.query.filter(Wdtest.name != 'wade').all()
​
# in 条件查询
user_list=Wdtest.query.filter(Wdtest.id.in_(['97124f50-0208-11ea-a66c-04ea56212bdf', '3'])).all()
​
# 所有⽤户先按年龄从⼩到⼤, 再按id从⼤到⼩排序, 取前5个
user_list=Wdtest.query.order_by(Wdtest.age,Wdtest.id.desc()).limit(5).all()
​
# 分⻚查询, 每⻚3个, 查询第2⻚的数据
pn = Wdtest.query.paginate(2,3)
print(pn.pages)
print(pn.page)
print(pn.items)

# 路由传递参数
@app.route('/user/<id>')
def user_info(id):
    return '%s' % id

@app.route('/test', methods=['GET', 'POST'])
def test():
    return "ok"

from werkzeug.routing import BaseConverter

# 自定义正则转换器
class RegexConverter(BaseConverter):
    def __init__(self, url_map, *args):
        super(RegexConverter, self).__init__(url_map)
        # 将接受的第1个参数当作匹配规则进行保存
        self.regex = args[0]

app = Flask(__name__)
​
# 将自定义转换器添加到转换器字典中，并指定转换器使用时名字为: regex
app.url_map.converters['regex'] = RegexConverter

@app.route('/index/<regex("[0-9]{3}"):id>')
def user_info(id):
    return "id 为 %s" % id

class RegexConverter(BaseConverter):
    def __init__(self, url_map, *args):
        super(RegexConverter, self).__init__(url_map)
        # 将接受的第1个参数当作匹配规则进行保存
        self.regex = args[0]
​
    def to_python(self, value):
        return int(value)

DEFAULT_CONVERTERS = {
    'default':          UnicodeConverter,
    'string':           UnicodeConverter,
    'any':              AnyConverter,
    'path':             PathConverter,
    'int':              IntegerConverter,
    'float':            FloatConverter,
    'uuid':             UUIDConverter,
}

# 配置日志
# LOG_LEVEL = "DEBUG"
LOG_LEVEL = "INFO"

FATAL/CRITICAL = 致命的，危险的
ERROR = 错误
WARNING = 警告
INFO = 信息
DEBUG = 调试
NOTSET = 没有设置

#增加日志模块
def setup_log(Config):
    #设置日志等级
    logging.basicConfig(level=Config.LOG_LEVEL)
    # 创建日志记录器，指明日志保存的路径、每个日志文件的最大大小、保存的日志文件个数上限
    file_log_handler=RotatingFileHandler('log/log',maxBytes=1024 * 1024 * 300, backupCount=10)
    # 创建日志记录的格式 日志等级 输入日志信息的文件名 行数 日志信息
    formatter = logging.Formatter('%(asctime)s: %(levelname)s %(filename)s:%(lineno)d %(message)s')
    # 为刚创建的日志记录器设置日志记录格式
    file_log_handler.setFormatter(formatter)
    # 为全局的日志工具对象（flaskapp使用的）添加日志记录器
    logging.getLogger().addHandler(file_log_handler)

#日志启动
setup_log(Config)

#新增redis连接模块
def connectRedis(Config):
    pool = redis.ConnectionPool(host=Config.REDIS_HOST, port=Config.REDIS_PORT, password=Config.REDIS_PASSWORD,
                                max_connections=Config.REDIS_POLL)
    redis_store = redis.Redis(connection_pool=pool)
    return redis_store

@index_blu.route("/redis",methods=["POST","GET"])
def add_toRedis():
    logging.info("come to here")
    key = request.args.get("key")
    application.redis_store.set(key , "1233")
    value=application.redis_store.get( key )
    print(value)
    return "12333"

# rabbitmq配置访问
# 添加用户名和密码
credentials = pika.PlainCredentials(Config.RABBITUSER, Config.RABBITPASSWORD)
# 配置连接参数
parameters = pika.ConnectionParameters(host=Config.RABBITHOST, port=Config.RABBITPORT, credentials=credentials)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

@index_blu.route("/rabitmq",methods=["POST","GET"])
def add_rabitmq():
    logging.info("come to rabiitmq")
    application.channel.queue_declare(queue='queuetest2')
​
    return "33333"

from flask import request
import application
​
# 拦截器,每次的请求进来都会做的操作
@application.app.before_request
def before_action():
    # 获取当前请求的路由（路径）
    a = request.path
    print(a)
    u = a.split('/')
    if len(a)>2:
        if u[1] == 'index':
            print('success')
    else:
        return "无权限请求"

#拦截器加载
requestFilter.before_action

from flask import request

class JSONResponse(Response):
​
    @classmethod
    def force_type(cls, response, environ=None):
        '''
        这个方法只有视图函数返回非字符、非元祖、非Response对象才会调用
        :param response:是视图函数的返回值
        :param environ:
        :return:
        '''
        print(response)
        print(type(response))
        if isinstance(response,(list,dict)):
​
            #jsonify除了将字典转换成json对象，还将对象包装成了一个Response对象
            response = jsonify(response)
​
        return super(JSONResponse,cls).force_type(response,environ) 
        
app.response_class = JSONResponse

@app.errorhandler(500)
def internal_server_error(e):
    return '服务器搬家了哈哈哈'
​
@app.errorhandler(404)
def internal_server_error(e):
    return '瞎请求什么路径呢'

@app.errorhandler(413)
def zero_division_error(e):
    return '除数不能为0'

@index_blu.route("/exception",methods=["POST","GET"])
def exception():
    logging.info("come to exception")
    try:
        print(2)
        a=3/0
    except:
        abort(413)
    return "ooooo"

current_app.name
current_app.test_value='value'

g.name='abc'

-c CONFIG : CONFIG,配置⽂件的路径，通过配置⽂件启动；⽣产环境使⽤； 
​
-b ADDRESS : ADDRESS，ip加端⼝，绑定运⾏的主机； 
​
-w INT, --workers INT：⽤于处理⼯作进程的数量，为正整数，默认为1； 
​
-k STRTING, --worker-class STRTING：要使⽤的⼯作模式，默认为sync异步，可以下载 
​
eventlet和gevent并指定 
​
--threads INT：处理请求的⼯作线程数，使⽤指定数量的线程运⾏每个worker。为正整数，默认为1。 
​
--worker-connections INT：最⼤客户端并发数量，默认情况下这个值为1000。 
​
--backlog int：未决连接的最⼤数量，即等待服务的客户的数量。默认2048个，⼀般不修改； 
​
-p FILE, --pid FILE：设置pid⽂件的⽂件名，如果不设置将不会创建pid⽂件 
​
--access-logfile FILE ： 要写⼊的访问⽇志⽬录--access-logformat STRING：要写⼊的访问⽇志格式 
​
--error-logfile FILE, --log-file FILE ： 要写⼊错误⽇志的⽂件⽬录。 
​
--log-level LEVEL ： 错误⽇志输出等级。 
​
--limit-request-line INT ： HTTP请求头的⾏数的最⼤⼤⼩，此参数⽤于限制HTTP请求⾏的允 
​
许⼤⼩，默认情况下，这个值为4094。值是0~8190的数字。 
​
--limit-request-fields INT ： 限制HTTP请求中请求头字段的数量。此字段⽤于限制请求头字 
​
段的数量以防⽌DDOS攻击，默认情况下，这个值为100，这个值不能超过32768 
​
--limit-request-field-size INT ： 限制HTTP请求中请求头的⼤⼩，默认情况下这个值为8190 
​
字节。值是⼀个整数或者0，当该值为0时，表示将对请求头⼤⼩不做限制 
​
-t INT, --timeout INT：超过这么多秒后⼯作将被杀掉，并重新启动。⼀般设定为30秒； 
​
--daemon： 是否以守护进程启动，默认false； 
​
--chdir： 在加载应⽤程序之前切换⽬录； 
​
--graceful-timeout INT：默认情况下，这个值为30，在超时(从接收到重启信号开始)之后仍然活着 
​
的⼯作将被强⾏杀死；⼀般使⽤默认； 
​
--keep-alive INT：在keep-alive连接上等待请求的秒数，默认情况下值为2。⼀般设定在1~5秒之 
​
间。 
​
--reload：默认为False。此设置⽤于开发，每当应⽤程序发⽣更改时，都会导致⼯作重新启动。 
​
--spew：打印服务器执⾏过的每⼀条语句，默认False。此选择为原⼦性的，即要么全部打印，要么全部 
​
不打印； 
​
--check-config ：显示现在的配置，默认值为False，即显示。 
​
-e ENV, --env ENV： 设置环境变量；