當前位置：首頁 > 编程语言 > python >内容正文

python

python flask 路由_python框架之Flask(2)-路由和视图Session

發布時間：2025/4/5 python 28 豆豆

生活随笔收集整理的這篇文章主要介紹了 python flask 路由_python框架之Flask(2)-路由和视图Session 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

路由和視圖

這一波主要是通過看源碼加深對 Flask 中路由和視圖的了解，可以先回顧一下裝飾器的知識：【裝飾器函數與進階】

路由設置的兩種方式

#示例代碼

from flask importFlask

app= Flask(__name__)

@app.route('/index')defindex():return 'index'

if __name__ == '__main__':

app.run()

直接看上面代碼，在?index?方法上通過裝飾器?@app.route('/index')?就建立路由?'/index'?和方法?index?的對應關系。

查看?app.route?的源碼：

1 def route(self, rule, **options):2 defdecorator(f):3 endpoint = options.pop('endpoint', None)4 self.add_url_rule(rule, endpoint, f, **options)5 returnf6

7 return decorator

flask.app.Flask.route

在上述示例中，?rule?就是我們自定義的路由參數?'/index'?；?endpoint?就是終結點參數(用來反向生成 url)，這里我們沒傳；而?f?實際上就是該裝飾器所裝飾的函數，在這里也就是?index?函數。其實到這里就可以斷定，該裝飾器實際上就是通過第?4?行的?add_url_rule?函數來建立路由和視圖的對應關系。我們可以測試：

from flask importFlask

app= Flask(__name__)defindex():return 'index'app.add_url_rule('/index', view_func=index)if __name__ == '__main__':

app.run()

在這里我去掉了?index?函數的裝飾器，而直接通過?app.add_url_rule?函數建立路由?'/index'?和?index?函數的對應關系，正常訪問。

endpoint

接著看?app.add_url_rule?函數做了什么：

1 @setupmethod2 def add_url_rule(self, rule, endpoint=None, view_func=None, provide_automatic_options=None, **options):3 if endpoint isNone:4 endpoint =_endpoint_from_view_func(view_func)5 options['endpoint'] =endpoint6 methods = options.pop('methods', None)7

8 if methods isNone:9 methods = getattr(view_func, 'methods', None) or ('GET',)10 ifisinstance(methods, string_types):11 raise TypeError('Allowed methods have to be iterables of strings,'

12 'for example: @app.route(..., methods=["POST"])')13 methods = set(item.upper() for item inmethods)14

15 required_methods = set(getattr(view_func, 'required_methods', ()))16

17 if provide_automatic_options isNone:18 provide_automatic_options =getattr(view_func,19 'provide_automatic_options', None)20

21 if provide_automatic_options isNone:22 if 'OPTIONS' not inmethods:23 provide_automatic_options =True24 required_methods.add('OPTIONS')25 else:26 provide_automatic_options =False27

28 methods |=required_methods29

30 rule = self.url_rule_class(rule, methods=methods, **options)31 rule.provide_automatic_options =provide_automatic_options32

33 self.url_map.add(rule)34 if view_func is notNone:35 old_func =self.view_functions.get(endpoint)36 if old_func is not None and old_func !=view_func:37 raise AssertionError('View function mapping is overwriting an'

38 'existing endpoint function: %s' %endpoint)39 self.view_functions[endpoint] = view_func

flask.app.Flask.add_url_rule

看?3、4、5?行，在示例中我們并沒有傳入?endpoint?參數，所以?endpoint?在第?3?行肯定是 None。接著執行第?4?行，查看?_endpoint_from_view_func?方法：

1 def_endpoint_from_view_func(view_func):2 assert view_func is not None, 'expected view func if endpoint'\3 'is not provided.'

4 return view_func.__name__

flask.helpers._endpoint_from_view_func

看第?4?行的返回值是視圖函數的函數名稱，所以當不傳?endpoint?參數時，?endpoint?的值就是視圖函數的函數名稱。

繼續看?flask.app.Flask.add_url_rule?函數的?34-39?行，?39?行做的就是每次裝飾器執行時，就會把裝飾器當前裝飾的函數當做值，?endpoint?當做?key?，放入?view_functions?這個字典中。而從?35-38?行可以看到，如果一個新的視圖函數的?endpoint?已經存在?view_functions?中，但這個函數又與?endpoint?之前對應的視圖函數不是同一個函數，就會產生?37?行錯誤。所以我們要保證每個視圖函數對應的?endpoint?不重復。

app.route的參數

除了我們已經熟悉的?rule?和?view_func?，它還可傳如下參數：

defaults

默認值，當URL中無參數，但函數需要參數時，可以使用 defaults={'k':'v'} 為函數提供參數。

endpoint

名稱，用于反向生成 URL，即：??url_for('endpoint')?。

methods

允許的請求方式，如：?methods=["GET","POST"]?。

strict_slashes

對 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

重定向到指定地址。

@app.route('/index/', redirect_to='/home/')defindex():return 'index'

subdomain

子域名訪問。

from flask importFlask

app= Flask(import_name=__name__)

app.config['SERVER_NAME'] = 'zze.com:5000'@app.route("/", subdomain="admin") #admin.zze.com:5000

defadmin_index():return "admin"@app.route("/", subdomain="guest") #guest.zze.com:5000

defguest_index():return "guest"@app.route("/dynamic", subdomain="") #http://test.zze.com:5000/dynamic

defdynamic_index(username):returnusernameif __name__ == '__main__':

app.run()

CBV

from flask importFlask, views

app= Flask(__name__)classTestView(views.MethodView):

methods= ['GET'] #只支持 GET 請求

decorators = [] #批量加上裝飾器

def get(self, *args, **kwargs):return 'GET'

def post(self, *args, **kwargs):return 'POST'app.add_url_rule('/test', None, TestView.as_view('test')) #as_view 的參數就是 endpoint

if __name__ == '__main__':

app.run()

它的實現其實和 Django 中的 CBV 實現很相似，源碼就不細說了。

正則匹配URL

from flask importFlask, views, url_forfrom werkzeug.routing importBaseConverter

app= Flask(import_name=__name__)#自定制類

classRegexConverter(BaseConverter):"""自定義URL匹配正則表達式"""

def __init__(self, map, regex):

super(RegexConverter, self).__init__(map)

self.regex=regexdefto_python(self, value):"""路由匹配時，匹配成功后傳遞給視圖函數中參數的值

:param value:

:return:"""

returnint(value)defto_url(self, value):"""使用url_for反向生成URL時，傳遞的參數經過該方法處理，返回的值用于生成URL中的參數

:param value:

:return:"""val=super(RegexConverter, self).to_url(value)returnval#注冊到 flask 的轉換器中

app.url_map.converters['regex'] =RegexConverter#使用

@app.route('/index/')defindex(nid):print(url_for('index', nid='888'))return 'Index'

if __name__ == '__main__':

app.run()

Session

源碼

首先我們要知道 Flask 初執行是會經過?flask.app.Flask.__call__?方法的，可以參考【Flask 的入口】。

def __call__(self, environ, start_response):#environ ：請求相關所有數據

#start_response ：用于設置響應相關數據

return self.wsgi_app(environ, start_response)

再查看?wsgi_app?方法：

1 defwsgi_app(self, environ, start_response):2 '''

3 獲取environ并對其進行封裝4 從environ中獲取名為session的cookie，解密并反序列化5 放入請求上下文6 '''

7 ctx =self.request_context(environ)8 error =None9 try:10 try:11 ctx.push()12 '''

13 執行視圖函數14 '''

15 response =self.full_dispatch_request()16 exceptException as e:17 error =e18 response =self.handle_exception(e)19 except:20 error = sys.exc_info()[1]21 raise

22 returnresponse(environ, start_response)23 finally:24 ifself.should_ignore_error(error):25 error =None26 '''

27 獲取session，解密并序列化，寫入cookie28 清空請求上下文29 '''

30 ctx.auto_pop(error)

flask.app.Flask.wsgi_app

environ?是請求相關信息，第?7?行將?environ?傳入?request_context?方法，看一下：

1 defrequest_context(self, environ):2 return RequestContext(self, environ)

flask.app.Flask.request_context

可以看到它的返回值就是以?environ?為構造參數傳入?RequestContext?類中的一個實例，看一下它初始化時做了什么：

1 def __init__(self, app, environ, request=None):2 self.app =app3 if request isNone:4 request =app.request_class(environ)5 self.request =request6 self.url_adapter =app.create_url_adapter(self.request)7 self.flashes =None8 self.session =None9

10 self._implicit_app_ctx_stack =[]11

12 self.preserved =False13

14 self._preserved_exc =None15

16 self._after_request_functions =[]17

18 self.match_request()

flask.ctx.RequestContext.__init__

看?3-8?行，?environ?傳入?request_class?方法中返回一個?request?實例，賦值給?self?，并在第?8?行給?self?新增一個?session?屬性并賦值為?None?。而最終這個?self?在?flask.app.Flask.wsgi_app?的第?7?行賦值給?ctx?。總結一下就是在執行完?flask.app.Flask.wsgi_app?的第?7?行后，?ctx?被賦值為?RequestContext?的一個實例，且這個實例中存在了將?environ?再次封裝的屬性?request?和一個為?None?的屬性?session?。

接著看到?flask.app.Flask.wsgi_app?中的第?11?行，查看?push?方法：

1 defpush(self):2 top =_request_ctx_stack.top3 if top is not None andtop.preserved:4 top.pop(top._preserved_exc)5

6 app_ctx =_app_ctx_stack.top7 if app_ctx is None or app_ctx.app !=self.app:8 app_ctx =self.app.app_context()9 app_ctx.push()10 self._implicit_app_ctx_stack.append(app_ctx)11 else:12 self._implicit_app_ctx_stack.append(None)13

14 if hasattr(sys, 'exc_clear'):15 sys.exc_clear()16

17 _request_ctx_stack.push(self)18

19 if self.session isNone:20 session_interface =self.app.session_interface21 self.session =session_interface.open_session(22 self.app, self.request23 )24

25 if self.session isNone:26 self.session = session_interface.make_null_session(self.app)

flask.ctx.RequestContext.push

直接看?19-26?行。如果?session?為空，就將傳入?app?和?request?參數執行?session_interface?的?open_session?方法的返回值賦給?session?，而此時這個?session_interface?默認就是?flask.sessions.SecureCookieSessionInterface?類的實例，查看其?open_session?方法：

1 defopen_session(self, app, request):2 s =self.get_signing_serializer(app)3 if s isNone:4 returnNone5 val =request.cookies.get(app.session_cookie_name)6 if notval:7 returnself.session_class()8 max_age =total_seconds(app.permanent_session_lifetime)9 try:10 data = s.loads(val, max_age=max_age)11 returnself.session_class(data)12 exceptBadSignature:13 return self.session_class()

flask.sessions.SecureCookieSessionInterface.open_session

看第?5?行是從?cookie?中取一個鍵為?app.session_cookie_name?的值，而這個鍵的默認值就是?'session'?，可在配置文件中配置(點擊查看配置文件默認配置參數)。緊接著就將這個值反序列化傳入?session_class?并返回?session_class?的實例，而?session_class?對應的是類?flask.sessions.SecureCookieSession?。所以在上面?flask.ctx.RequestContext.push?方法中?21?行?self.session?的值就是?flask.sessions.SecureCookieSession?的實例。也就是在上面?flask.app.Flask.wsgi_app?的?11?行執行之后，?ctx?的?session?就有值了。

接著看?flask.app.Flask.wsgi_app?的第?15?行，這行就是通過?full_dispatch_request?方法來完成執行視圖函數和部分請求的收尾操作：

1 deffull_dispatch_request(self):2 self.try_trigger_before_first_request_functions()3 try:4 request_started.send(self)5 rv =self.preprocess_request()6 if rv isNone:7 rv =self.dispatch_request()8 exceptException as e:9 rv =self.handle_user_exception(e)10 return self.finalize_request(rv)

flask.app.Flask.full_dispatch_request

看第?10?行，在完成了上面視圖函數相關操作后，通過?finalize_request?方法完成請求收尾操作：

1 def finalize_request(self, rv, from_error_handler=False):2 response =self.make_response(rv)3 try:4 response =self.process_response(response)5 request_finished.send(self, response=response)6 exceptException:7 if notfrom_error_handler:8 raise

9 self.logger.exception('Request finalizing failed with an'

10 'error while handling an error')11 return response

flask.app.Flask.finalize_request

再看到第?4?行的?process_response?方法，這個方法如其名，就是用來處理響應相關信息：

1 defprocess_response(self, response):2 ctx =_request_ctx_stack.top3 bp =ctx.request.blueprint4 funcs =ctx._after_request_functions5 if bp is not None and bp inself.after_request_funcs:6 funcs =chain(funcs, reversed(self.after_request_funcs[bp]))7 if None inself.after_request_funcs:8 funcs =chain(funcs, reversed(self.after_request_funcs[None]))9 for handler infuncs:10 response =handler(response)11 if notself.session_interface.is_null_session(ctx.session):12 self.session_interface.save_session(self, ctx.session, response)13 return response

flask.app.process_response

直接看?11、12?行，當?session?不為空時，調用?session_interface.save_session?方法，而?session_interface?就是上面執行?open_session?方法的?flask.sessions.SecureCookieSessionInterface?類實例。

1 defsave_session(self, app, session, response):2 domain =self.get_cookie_domain(app)3 path =self.get_cookie_path(app)4

5 if notsession:6 ifsession.modified:7 response.delete_cookie(8 app.session_cookie_name,9 domain=domain,10 path=path11 )12

13 return

15 ifsession.accessed:16 response.vary.add('Cookie')17

18 if notself.should_set_cookie(app, session):19 return

21 httponly =self.get_cookie_httponly(app)22 secure =self.get_cookie_secure(app)23 samesite =self.get_cookie_samesite(app)24 expires =self.get_expiration_time(app, session)25 val =self.get_signing_serializer(app).dumps(dict(session))26 response.set_cookie(27 app.session_cookie_name,28 val,29 expires=expires,30 httponly=httponly,31 domain=domain,32 path=path,33 secure=secure,34 samesite=samesite35 )

flask.sessions.SecureCookieSessionInterface.save_session

看?25-35?行，會發現最后又將?session?序列化，再次寫入 cookie 。

得出結論：當請求剛到來時，flask 讀取 cookie 中 session 對應的值，并將該值解密并反序列化成字典，放入內存以便視圖函數使用；當請求結束時，flask 會讀取內存中字典的值，進行序列化和加密，再寫入到 cookie 中。

第三方session

使用

1 from flask importFlask, request, session, redirect2 from flask.sessions importSecureCookieSessionInterface3 from flask_session importSession4 from redis importRedis5

6 app = Flask(__name__)7 app.debug =True8

10 app.config['SESSION_REDIS'] = Redis(host='127.0.0.1', port='6379', password='1234')11 #設置 session 類型

12 app.config['SESSION_TYPE'] = 'redis'

13 #設置根據 session 類型設置 app.session_interface

14 Session(app)15

17 @app.route('/login')18 deflogin():19 session['username'] = 'zze'

20 return 'success'

23 app.run()

源碼

通過上面的源碼部分已經知道了，flask 中的 session 存取就是通過?app.session_interface?來完成的，默認?app.session_interface = SecureCookieSessionInterface()?，而我們只要修改這一部分，讓其存取是通過 redis 就 ok 了。查看?14?行?Session?類：

1 classSession(object):2 def __init__(self, app=None):3 self.app =app4 if app is notNone:5 self.init_app(app)6

7 definit_app(self, app):8 app.session_interface =self._get_interface(app)9

10 def_get_interface(self, app):11 config =app.config.copy()12 config.setdefault('SESSION_TYPE', 'null')13 config.setdefault('SESSION_PERMANENT', True)14 config.setdefault('SESSION_USE_SIGNER', False)15 config.setdefault('SESSION_KEY_PREFIX', 'session:')16 config.setdefault('SESSION_REDIS', None)17 config.setdefault('SESSION_MEMCACHED', None)18 config.setdefault('SESSION_FILE_DIR',19 os.path.join(os.getcwd(), 'flask_session'))20 config.setdefault('SESSION_FILE_THRESHOLD', 500)21 config.setdefault('SESSION_FILE_MODE', 384)22 config.setdefault('SESSION_MONGODB', None)23 config.setdefault('SESSION_MONGODB_DB', 'flask_session')24 config.setdefault('SESSION_MONGODB_COLLECT', 'sessions')25 config.setdefault('SESSION_SQLALCHEMY', None)26 config.setdefault('SESSION_SQLALCHEMY_TABLE', 'sessions')27

28 if config['SESSION_TYPE'] == 'redis':29 session_interface =RedisSessionInterface(30 config['SESSION_REDIS'], config['SESSION_KEY_PREFIX'],31 config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])32 elif config['SESSION_TYPE'] == 'memcached':33 session_interface =MemcachedSessionInterface(34 config['SESSION_MEMCACHED'], config['SESSION_KEY_PREFIX'],35 config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])36 elif config['SESSION_TYPE'] == 'filesystem':37 session_interface =FileSystemSessionInterface(38 config['SESSION_FILE_DIR'], config['SESSION_FILE_THRESHOLD'],39 config['SESSION_FILE_MODE'], config['SESSION_KEY_PREFIX'],40 config['SESSION_USE_SIGNER'], config['SESSION_PERMANENT'])41 elif config['SESSION_TYPE'] == 'mongodb':42 session_interface =MongoDBSessionInterface(43 config['SESSION_MONGODB'], config['SESSION_MONGODB_DB'],44 config['SESSION_MONGODB_COLLECT'],45 config['SESSION_KEY_PREFIX'], config['SESSION_USE_SIGNER'],46 config['SESSION_PERMANENT'])47 elif config['SESSION_TYPE'] == 'sqlalchemy':48 session_interface =SqlAlchemySessionInterface(49 app, config['SESSION_SQLALCHEMY'],50 config['SESSION_SQLALCHEMY_TABLE'],51 config['SESSION_KEY_PREFIX'], config['SESSION_USE_SIGNER'],52 config['SESSION_PERMANENT'])53 else:54 session_interface =NullSessionInterface()55

56 return session_interface

flask_session.Session

執行到第?8?行，可以看到它就是給?app.session_interface?賦值為?self._get_interface(app)?，而這個方法的返回值是根據在上面使用中第?12?行配置的?'SESSION_TYPE'?字段決定的，這里設置的是?'redis'?，所以?self._get_interface(app)?的返回值就為第?82?行的?RedisSessionInterface?實例。

總結

以上是生活随笔為你收集整理的python flask 路由_python框架之Flask(2)-路由和视图Session的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：红米note5解锁教程_红米NOTE5如
下一篇： shell中join链接多个域_Linu