flask（精讲）
Flask
一：web框架Django和Flask本質
socket服務端
?| 1 2 3 | wsgi： Web服務網關接口 ????- wsgiref?????????? # Django內部內置模塊 ????- werkzeug????????? # Flask安裝完成后，內部默認已經安裝好werkzeug |
?
二：簡單的Flask
創建Flask s1，生成最簡單的代碼。運行s1.py文件，flask運行成功。
from flask import Flaskapp = Flask(__name__)# 路由映射關系 @app.route('/') def hello_world():return 'Hello World!'if __name__ == '__main__':app.run() s1.py?
三：配置
?以下配置參數為app = Flask(__name__)的參數，查看源碼類Flask __init__中可傳的參數
import_name, # 就是Flask(__name__)中的__name__，一般寫__name__ static_path=None, # 靜態文件路徑，這個即將被廢棄了 static_url_path=None, # 靜態前綴：static_url_path = '/sssss'。創建flask時目錄被默認創建為/static,未配置該參數時，訪問127.0.0.1:5000/static/1.jpg就可訪問/static目錄下的圖片但是修改配置后直接訪問127.0.0.1:5000/sssss/1.jpg就可訪問/static目錄下的圖片 static_folder='static', # 靜態文件目錄，創建Flask時目錄/static被默認創建 template_folder='templates', # 模板路徑，創建Flask時目錄/templates被默認創建。from flask import Flask,render_template return render_template('hello.html') instance_path=None, # C:\Users\Administrator\PycharmProjects\s133\instance，用的少，默認是路徑，當前目錄 + \instance instance_relative_config=False, # 當為True，會默認去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件。如果為Flase時，不管它。 root_path=None # C:\Users\Administrator\PycharmProjects\s133，當前目錄。默認在當前目錄找配置文件instance_relative_config=True時，默認去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件 View Code?以下配置為flask.config.Config對象（繼承字典）的默認參數
{'DEBUG': get_debug_flag(default=False), # 是否開啟Debug模式'TESTING': False, # 是否開啟測試模式'PROPAGATE_EXCEPTIONS': None, 'PRESERVE_CONTEXT_ON_EXCEPTION': None,'SECRET_KEY': None,'PERMANENT_SESSION_LIFETIME': timedelta(days=31), # session的超時時間'USE_X_SENDFILE': False,'LOGGER_NAME': None,'LOGGER_HANDLER_POLICY': 'always','SERVER_NAME': None,'APPLICATION_ROOT': None,'SESSION_COOKIE_NAME': 'session','SESSION_COOKIE_DOMAIN': None,'SESSION_COOKIE_PATH': None,'SESSION_COOKIE_HTTPONLY': True,'SESSION_COOKIE_SECURE': False,'SESSION_REFRESH_EACH_REQUEST': True,'MAX_CONTENT_LENGTH': None,'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12),'TRAP_BAD_REQUEST_ERRORS': False,'TRAP_HTTP_EXCEPTIONS': False,'EXPLAIN_TEMPLATE_LOADING': False,'PREFERRED_URL_SCHEME': 'http','JSON_AS_ASCII': True,'JSON_SORT_KEYS': True,'JSONIFY_PRETTYPRINT_REGULAR': True,'JSONIFY_MIMETYPE': 'application/json','TEMPLATES_AUTO_RELOAD': None, } View Code app.config['DEBUG'] = True # 進入調試模式 app.debug = True # 進入調試模式 app.session_interface # session的接口 app.config.updata({}) 配置方式一：(s1.py中通過操作字典的方式) 第一種：去一個.py文件中導入配置,例如flask目錄下創建一個settings.py，與staic目錄同一級別s133.py:app.config.from_pyfile("settings.py")settings.py:DEBUG = True第二種：環境變量中取app.config.from_envvar("環境變量名稱"),內部調用from_pyfile方法使用：test.py:import osos.environ['xxxxx'] = "settings" # 或者os.environ['xxxxx'] = "settings.py"，settings加入環境變量 s133.py:app.config.from_envvar("xxxxx") # 找到settings對象，然后執行第一種app.config.from_pyfile("settings.py") 第三種：同第一種方式，創建json.py文件,s133.py中調用from_json方法 · app.config.from_json("json文件名稱")JSON文件名稱，必須是json格式，因為內部會執行json.loads第四種：字典的格式app.config.from_mapping({'DEBUG':True})第五種：比較推薦使用的，注意要寫大寫，小寫是導入不成功的。app.config.from_object("settings.TestingConfig")settings.py：class Config(object):DEBUG = FalseTESTING = FalseDATABASE_URI = 'sqlite://:memory:'class ProductionConfig(Config):DATABASE_URI = 'mysql://user@localhost/foo'class DevelopmentConfig(Config):DEBUG = Trueclass TestingConfig(Config):TESTING = True 配置方式二?
四：路由
路由使用：
@app.route('/') def hello_world():return 'Hello World!' 方法一：裝飾器方式 def hello_world():# 反向生成urlfrom flask import url_forurl = url_for('xxx') # url此時為 / return 'Hello World!' app.add_url_rule('/',view_func=hello_world,endpoint='xxx',methods=["GET","POST"]) # view_func視圖函數；endpoint和django中的name一樣，反向生成url，不加endpoint，endpoint默認值為視圖函數名 方式二：url正則匹配：
@app.route('/edit/<int:nid>') def hello_world(nid):return 'Hello World!' 示例 @app.route('/user/<username>') @app.route('/post/<int:post_id>') @app.route('/post/<float:post_id>') @app.route('/post/<path:path>') @app.route('/login', methods=['GET', 'POST']) 常用的路由系統,django支持自己寫正則表達式,flask不支持 DEFAULT_CONVERTERS = {'default': UnicodeConverter,'string': UnicodeConverter,'any': AnyConverter,'path': PathConverter,'int': IntegerConverter,'float': FloatConverter,'uuid': UUIDConverter, } 所有的路由系統都是基于對應關系來處理 from flask import Flask, views, url_for from werkzeug.routing import BaseConverterapp = Flask(import_name=__name__)class RegexConverter(BaseConverter):"""自定義URL匹配正則表達式"""def __init__(self, map, regex):super(RegexConverter, self).__init__(map)self.regex = regexdef to_python(self, value):"""路由匹配時，匹配成功后傳遞給視圖函數中參數的值:param value: :return: """return int(value)def to_url(self, value):"""使用url_for反向生成URL時，傳遞的參數經過該方法處理，返回的值用于生成URL中的參數:param value: :return: """val = super(RegexConverter, self).to_url(value)return val+'666'# 添加到flask中 app.url_map.converters['regex'] = RegexConverter# 自定義的url正則的使用 @app.route('/index/<regex("\d+"):nid>') def index(nid):print(url_for('index', nid='888')) # 反向生成url /index/888666/ ,反向生成url之前會先執行to_url方法return 'Index'if __name__ == '__main__':app.run() 自定制url正則匹配 方法一：def auth(func):def inner(*args, **kwargs):print('before')result = func(*args, **kwargs)print('after')return resultreturn inner@app.route('/index.html',methods=['GET','POST'],endpoint='index')@authdef index():return 'Index'方法二：def auth(func):def inner(*args, **kwargs):print('before')result = func(*args, **kwargs)print('after')return resultreturn inner class IndexView(views.MethodView):methods = ['GET']decorators = [auth, ] # 執行的裝飾器def get(self):return 'Index.GET'def post(self):return 'Index.POST'app.add_url_rule('/index', view_func=IndexView.as_view(name='index')) # name=endpoint flask中裝飾器的使用 rule, URL規則 view_func, 視圖函數名稱 defaults=None, 默認值,當URL中無參數，函數需要參數時，使用defaults={'nid':9}為函數提供參數 endpoint=None, 名稱，用于反向生成URL，即： url_for('名稱') methods=None, 允許的請求方式，如：["GET","POST"]strict_slashes=None, 對URL最后的 / 符號是否嚴格要求，如：@app.route('/index',strict_slashes=False)，訪問 http://www.xx.com/index/ 或 http://www.xx.com/index均可@app.route('/index',strict_slashes=True)僅訪問 http://www.xx.com/index redirect_to=None, 重定向到指定地址如：@app.route('/index/<int:nid>', redirect_to='/home/<nid>') # 請求到來不執行/index/<int:nid>代碼，直接重定向到/home/<nid> 或def func(adapter, nid):return "/home/888"@app.route('/index/<int:nid>', redirect_to=func)subdomain=None, 子域名訪問如：from flask import Flask, views, url_forapp = Flask(import_name=__name__)app.config['SERVER_NAME'] = 'xuyaping.com:5000' # 必須寫，才能支持子域名 @app.route("/index/", subdomain="admin") # 訪問http://admin/xuyaping.com:5000/index/ def static_index():"""Flask supports static subdomainsThis is available at static.your-domain.tld"""return "static.your-domain.tld"if __name__ == '__main__':app.run() @app.route和app.add_url_rule參數?
五：模板
模板的使用
Flask使用的是Jinja2模板，所以其語法和Django無差別
不過在django模板中執行函數或方法時，不用加括號就會自己執行，而Flask必須自己加括號才會執行。
flask中的Markup等價django的mark_safe
自定義模板方法
創建一個函數并通過參數的形式傳入render_template，如：
?
六：請求和響應
from flask import Flask from flask import request from flask import render_template from flask import redirect from flask import make_responseapp = Flask(__name__)@app.route('/login.html', methods=['GET', "POST"]) def login():# 請求相關信息# request.method# request.args # GET傳的參數# request.form # 表單，POST傳的參數# request.values# request.cookies# request.headers# request.path# request.full_path# request.script_root# request.url# request.base_url# request.url_root# request.host_url# request.host# request.files # 文件# obj = request.files['the_file_name']# obj.save('/var/www/uploads/') # save直接把文件存儲到/var/www/uploads/目錄中了# 響應相關信息# return "字符串" # 相當于django中的Httpresponse# return render_template('html模板路徑',**{}) # 相當于django中的render# return redirect('/index.html') # 相當于django中的redirect# response = make_response(render_template('index.html')) # make_response把返回的數據封裝起來，然后就有了delete_cookie、set_cookie、headers方法了# response是flask.wrappers.Response類型# response.delete_cookie('key')# response.set_cookie('key', 'value')# response.headers['X-Something'] = 'A value'# return responsereturn "內容"if __name__ == '__main__':app.run() View Code?
七：session
?| 1 2 3 | flask內置session默認放在加密Cookie中，依賴于session.secret_key?? 設置：session['username'] ＝ 'xxx' 刪除：session.pop('username', None) |
自定義session及使用
import uuid import json from flask.sessions import SessionInterface from flask.sessions import SessionMixin from itsdangerous import Signer, BadSignature, want_bytesclass MySession(dict, SessionMixin):def __init__(self, initial=None, sid=None):self.sid = sidself.initial = initialsuper(MySession, self).__init__(initial or ())def __setitem__(self, key, value):super(MySession, self).__setitem__(key, value)def __getitem__(self, item):return super(MySession, self).__getitem__(item)def __delitem__(self, key):super(MySession, self).__delitem__(key)class MySessionInterface(SessionInterface):session_class = MySessioncontainer = {}def __init__(self):import redisself.redis = redis.Redis()def _generate_sid(self):return str(uuid.uuid4())def _get_signer(self, app):if not app.secret_key:return Nonereturn Signer(app.secret_key, salt='flask-session',key_derivation='hmac')def open_session(self, app, request):"""程序剛啟動時執行，需要返回一個session對象"""sid = request.cookies.get(app.session_cookie_name)if not sid:sid = self._generate_sid()return self.session_class(sid=sid)signer = self._get_signer(app)try:sid_as_bytes = signer.unsign(sid)sid = sid_as_bytes.decode()except BadSignature:sid = self._generate_sid()return self.session_class(sid=sid)# session保存在redis中# val = self.redis.get(sid)# session保存在內存中val = self.container.get(sid)if val is not None:try:data = json.loads(val)return self.session_class(data, sid=sid)except:return self.session_class(sid=sid)return self.session_class(sid=sid)def save_session(self, app, session, response):"""程序結束前執行，可以保存session中所有的值如：保存到resit寫入到用戶cookie"""domain = self.get_cookie_domain(app)path = self.get_cookie_path(app)httponly = self.get_cookie_httponly(app)secure = self.get_cookie_secure(app)expires = self.get_expiration_time(app, session)val = json.dumps(dict(session))# session保存在redis中# self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime)# session保存在內存中 self.container.setdefault(session.sid, val)session_id = self._get_signer(app).sign(want_bytes(session.sid))response.set_cookie(app.session_cookie_name, session_id,expires=expires, httponly=httponly,domain=domain, path=path, secure=secure) sessions.py from sessions import MySessionInterface app.session_interface = MySessionInterface() 使用或者使用flask-session模塊，配置文件中設置
from flask import Flask from flask import session from pro_flask.utils.session import MySessionInterface app = Flask(__name__)app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT' app.session_interface = MySessionInterface()@app.route('/login.html', methods=['GET', "POST"]) def login():print(session)session['user1'] = 'alex'session['user2'] = 'alex'del session['user2']return "內容"if __name__ == '__main__':app.run() 使用flask-session?
八：message
message是一個基于Session實現的用于保存數據的集合，其特點是：使用一次就刪除。
from flask import Flask, flash, redirect, render_template, request, get_flashed_messagesapp = Flask(__name__) app.secret_key = 'some_secret'@app.route('/') def index1():messages = get_flashed_messages() # 從session中取，取到就刪掉print(messages)return "Index1"@app.route('/set') def index2():v = request.args.get('p')flash(v) # 存儲在session中return 'ok'if __name__ == "__main__":app.run() View Code?
九：擴展：偽中間件
from flask import Flask, flash, requestapp = Flask(__name__) app.secret_key = 'some_secret'@app.route('/index') def index():return 'index.html'# 中間件 class MiddleWare:def __init__(self,wsgi_app):self.wsgi_app = wsgi_appdef __call__(self, environ, start_response): # environ, start_response是wsgi socket傳的參數print('before')response = self.wsgi_app(environ, start_response)print('after')return responseif __name__ == "__main__":app.wsgi_app = MiddleWare(app.wsgi_app)app.run(port=9999) View Code?
十：Flask插件
?| 1 2 3 | WTForms??? ??? form組件，做form表單驗證的組件 SQLAchemy ??? ORM操作 Flask-Session session插件 |
十一：藍圖
藍圖的功能就是將不同功能放在不同的py文件中
eg：
order.py
+ View Code?| 1 2 3 4 5 6 | from flask import Blueprint order = Blueprint('order',__name__) @order.route('/order') def order(): ????return 'Order' |
account.py
+ View Code?| 1 2 3 4 5 6 | from flask import Blueprint,render_template account = Blueprint('account',__name__) @account.route('/login') def login(): ????return render_template('login.html') |
_init_.py
+ View Code?| 1 2 3 4 5 6 7 | from flask import Flask from .views import account from .views import order app = Flask(__name__) app.register_blueprint(account.account) app.register_blueprint(order.order) |
?十二：數據庫連接池
?
""" 為每個線程創建一個連接，thread.local實現。"""from DBUtils.PersistentDB import PersistentDB import pymysqlPOOL = PersistentDB(creator=pymysql, # 使用鏈接數據庫的模塊maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayscloseable=False,# 如果為False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，才會自動關閉鏈接。如果為True時， conn.close()則關閉鏈接，那么再次調用pool.connection時就會報錯，因為已經真的關閉了連接（pool.steady_connection()可以獲取一個新的鏈接）threadlocal=None, # 本線程獨享值得對象，用于保存鏈接對象，如果鏈接對象被重置host='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8' )def func():# conn = SteadyDBConnection()conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()cursor.close()conn.close() # 不是真的關閉，而是假的關閉。 conn = pymysql.connect() conn.close() conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()cursor.close()conn.close()import threadingfor i in range(10):t = threading.Thread(target=func)t.start() 模式一 import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB(creator=pymysql, # 使用鏈接數據庫的模塊maxconnections=6, # 連接池允許的最大連接數，0和None表示不限制連接數mincached=2, # 初始化時，鏈接池中至少創建的空閑的鏈接，0表示不創建 maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制maxshared=3, # 鏈接池中最多共享的鏈接數量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8' )def func():# 檢測當前正在運行連接數的是否小于最大鏈接數，如果不小于則：等待或報raise TooManyConnections異常# 否則# 則優先去初始化時創建的鏈接中獲取鏈接 SteadyDBConnection。# 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。# 如果最開始創建的鏈接沒有鏈接，則去創建一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中并返回。# 一旦關閉鏈接后，連接就返回到連接池讓后續線程繼續使用。# PooledDedicatedDBConnectionconn = POOL.connection()# print(th, '鏈接被拿走了', conn1._con)# print(th, '池子里目前有', pool._idle_cache, '\r\n') cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()func() 模式二?
?
轉載于:https://www.cnblogs.com/liuchengdong/p/8269606.html
總結
- 上一篇： 生产质量分析报告常用模板汇总，质量人必备
- 下一篇： uefiboot 文件_UEFI启动文件