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： 設定環境變數；