web 也叫CS開發
CS 及客戶端,服務器端編程
客戶端,服務器端之間須要socket,約定協議,版本(每每使用的協議是TCP或UDP),指定地址和端口,就能夠通訊了html客戶端,服務端傳輸數據,數據能夠有必定的格式,雙方必須約定好 前端
B=Browser,Browser是一種特殊的客戶端,支持HTTP(S)協議,可以經過URL 向服務器端發起請求,等待服務端返回HTML等數據,並在瀏覽器內可視化展現程序 python
S=server,Server支持HTTP(S)協議,可以接受衆多客戶端發起的HTTP請求,通過處理,將HTML等數據返回給瀏覽器 mysql
本質上來講,BS是一種特殊的CS,及客戶端必須是一種支持HTTP協議且可以解析並渲染HTML的軟件,服務端必須是可以接受客戶端HTTP訪問的服務器軟件 程序員
HTTP 底層使用TCP傳輸,須要經過相應的規則來處理web
HTTP 超文本傳輸協議,在文本的基礎上作一些突破,相關的渲染處理,斷行等。使用標籤的方式進行規定的處理,文本已經有了一個格式,瀏覽器中必須一種能力,支持這種文本的處理和渲染。正則表達式
瀏覽器必須支持HTTP協議,必須可以理解超文本並將其繪製出來 redis
BS 開發分爲兩端開發
1 客戶端開發,或稱爲前端開發
2 服務端開發,python能夠學習WSGI,Django,Flask,Tornado等sqlpython WEB 框架
WSGI, web Server Gateway interface,能夠看作是一種底層協議,它規定了服務器程序和應用程序各自實現什麼藉口,python稱爲wsgiref 數據庫flask: 基於WSGI ,微框架
Django:基於WSGI,開源的WEB框架
在http1.1以前,都是一個請求一個鏈接,而TCP的連接建立銷燬成本高,對服務器影響較大,所以自從http1.1開始,支持keep-alive,默認也開啓,一個鏈接打開後,會保持一段時間,瀏覽器再訪問該服務器資源就使用這個TCP連接,減輕了服務器的壓力,提升了效率
全部的動態網頁開發,都必須是有狀態的協議
若是是持續鏈接,一個TCP是能夠發送多個HTTP請求的
HTTP/1.1存在一個問題,單個TCP鏈接在同一時刻只能處理一個請求,意思是說: 兩個請求的生命週期不能重疊,任意兩個HTTP請求從開始到結束的時間在同一個TCP鏈接裏不能重疊
雖然HTTP/1.1規範中規定了Pipelining來試圖解決這個問題,但此功能默認是關閉的
Pipelining 中
客戶端能夠在一個鏈接中發送多個請求(不須要等待任意請求的響應)。收到請求的服務器必須按照請求收到的順序發送響應。pipelining的缺點
1 一些代理服務器不能正確支持 HTTP pipelining
2 正確的流水線實現是複雜的
3 若是第一個請求的處理花費大量時間,則會致使後面的請求沒法處理,形成阻塞。
Http2 提供了Multiplexing 多路傳輸特性,能夠在一個TCP鏈接中同時完成多個HTTP請求
1 維持和服務其已經創建的TCP鏈接,在同一個鏈接上順序處理多個請求
2 和服務器創建多個TCP鏈接瀏覽器對同一個Host 創建TCP鏈接數量有沒限制
Chrome 最多容許對同一個Host創建6個TCP連接,不一樣瀏覽器有區別
若是圖片都是HTTPS 鏈接而且在同一域名下,那麼瀏覽器在SSL握手以後會和服務器協商能不能使用HTTP2,若是能的話就是用Multiplexing 功能在這個鏈接上進行多路傳輸,不過也未必會全部掛載在各個域名的資源都會使用一個TCP鏈接獲取嗎,但能夠肯定的是multiplexing 可能很被用到
若是發現不是使用HTTP2,或者不用HTTPS,(現實中的 HTTP2 都是在 HTTPS 上實現的,因此也就是隻能使用 HTTP/1.1),那麼瀏覽器就會在一個HOST上創建多個TCP鏈接,鏈接數量的最大限制取決於瀏覽器的設置,這些來凝結會在空閒的時候被瀏覽器用來發送新請求,若是全部鏈接都在發送請求,那麼其只能等待了。
同一個客戶端的兩次請求之間沒有任何關係,從服務端的角度看,他不知道這兩個請求來自同一個客戶端
最先的設計是不須要知道二者之間的聯繫的,
HTTP協議是無狀態協議
有連接,由於HTTP 是基於TCP 連接的,須要3次握手,4次斷開
http://127.0.0.1/login?user=zhangsan&password=123
登陸窗口不能使用GET傳輸,GET頭部的長度是有限的,不能多於200多個之外的傳輸
格式是 ? 後面加key1=value1&key2=value2
當使用POST 傳輸數據時,其相關的數據都被封裝在請求體及body中,而不是get中的直接暴露。
大的數據傳輸,必須使用POST,而不能使用GET傳輸數據。
HTML 是一種格式的約定,須要的數據是動態的,去數據庫查的數據不是死的,是動態的,靜態文本文件包括圖片
HTML 是將文本原封不動的返回,如果一個登錄的用戶名和密碼的匹配問題的時候,就不是HTML能作的事情,此時便須要動態網頁來完成。如python,只有腳本是不行的,這就須要相似的解釋器來進行處理。Php,asp等動態的網頁技術,server page 服務器端的頁面。動態頁面中的無狀態帶來很大的問題,再次登陸將致使登陸後的和登陸的不要緊。既然你連接到我,我能夠發送一個惟一標識給你,你須要下次將這個標識帶來,來保證是你,服務端須要發送和記錄標識,此處須要寫入到內存的數據結構中,當用戶量很大時,記錄的東西就不只僅是這個用戶標識了。
cookie:是一種客戶端,服務端傳遞數據的技術 ,其保存的形式是鍵值對信息
瀏覽器發起每個請求,都會把cookie信息給服務端,服務端能夠經過判斷這些信息,來肯定此次請求是否和以前的請求有關聯
通常來講cookie信息是在服務器端生成,返回給客戶端
客戶端能夠本身設置cookie信息
Cookie 通常是當你第一次連接服務器的時候服務器會查看是否有cookie帶過來,若沒有則推送一個標識,這個標識中會在HTTP的response包中存在,其會在瀏覽器中保存起來。若是再次對一樣網站發起請求,若是cookie沒過時時,其會繼續處理此標識。如果同一個且有效,則若登陸過,則不顯示登陸頁面,若沒登陸,則強制跳轉到登陸頁面。若是一個網站一直登陸,其發現cookie快過時了,則會延長。
Cookie 是對不一樣的域名有區分的
cookie中加的ID 叫作session ID ,稱爲會話ID,當會話完結後,ID就消亡了,瀏覽器關閉,
Session 是存放在服務器端的,其會增長內存。後期則使用無session, token每每中間會使用redis和memcached進行處理
請求來的時候,其得帶着是不是同一個會話標識
cookie能夠僞造
WSGI 主要規定了服務器端和應用程序之間的接口
瀏覽器
能夠接受用戶的socket請求並和客戶端達成HTTP協議並識別解析,將數據交給後端的WSGI app 進行處理
Server 必須支持HTTP協議,在python中實現了WSGI的接口,HTTP server得支持WSGI協議,將數據傳遞給程序,(app返回)而後返回給客戶端對應的狀態狀況(響應頭),使得瀏覽器作好準備,而後再返回給server,再由server將其包裝成HTTP的協議並解析處理。
後端真實處理業務的函數對象
後端APP知足的條件
1 可經過前面的WGSI Server進行相關的調用操做
應用程序應該是一個可調用對象
調用實際上是回調,調用的實際上是APP的某個方法
python中應該是函數,類,實現了call方法的類的實例
2 這個可調用對象應該接受兩個參數
知足了WSGI 的基本要求,必須再留一個空,協議的封裝是須要在server端的,所以要將你寫的東西交給 http server ,由http server對返回結果進行處理 其上述返回必須是一個可迭代對象(list,dict等)兩個參數就是入 request和出response
Handler 和 body都給了app
邏輯處理: 調用對應的方法給客戶端。
http server 返回給app server 的參數
eviron和start_response 這兩個參數能夠是任意的合法名。但通常都是這兩個名字
eviron 是包含HTTP請求信息的dict對象
名稱 | 含義 |
---|---|
REQUEST_METHOD | 請求方法,GET,PSOT,HEAD等 |
PATH_INFO | URL 中路徑部分信息 |
QUERY_STRING | 查詢字符串 |
SERVER_NAME,SERVER_PORT | 服務器名,端口號 |
HTTP_POST | 地址和端口 |
SERVER_PROTOCOL | 協議 |
HTTP_USER_AGENT | User Agent信息 |
start_response 是一個可調用對象,有3個參數,定義以下:
start_response(status,response_headers,exc_info=None)
status 是狀態碼。如200 ok
response_headers 是一個元素爲二元祖的列表,如[('Content-Type','text/plain;charset=utf-8')]
exec_info 在錯誤處理的時候使用
start_response 應該在返回可迭代對象以前調用,由於他返回的是Response Header,返回的可迭代對象是Response Body。
先發頭部,而後纔是body
服務器端
服務器端程序須要調用符合上述定義的可調用對象,傳入environ,start_response拿到返回可迭代對象,返回給客戶端。
WSGIREF 是一個WSGI 的參考實現庫
wsgiref.simple_server 實現了一個簡單的WSGI HTTP服務器
相關參數以下
wsgiref.simple_server.make_server(
host, port, app, server_class=WSGIServer, handler_class=WSGIRequestHandler
)源碼以下
def make_server( host, port, app, server_class=WSGIServer, handler_class=WSGIRequestHandler ): """Create a new WSGI server listening on `host` and `port` for `app`""" server = server_class((host, port), handler_class) server.set_app(app) return server
經過demo app 實現基本的展現頁面
def demo_app(environ,start_response): from io import StringIO stdout = StringIO() print("Hello world!", file=stdout) print(file=stdout) h = sorted(environ.items()) for k,v in h: print(k,'=',repr(v), file=stdout) start_response("200 OK", [('Content-Type','text/plain; charset=utf-8')]) return [stdout.getvalue().encode("utf-8")]
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server,demo_app ip='192.168.1.200' port=80 server=make_server(ip,port,demo_app) # 實例化一個websever server.serve_forever() # 啓動 server.server_close() # 關閉 server.shutdown() # 刪除
General Request URL: http://192.168.1.200/ Request Method: GET Status Code: 200 OK Remote Address: 192.168.1.200:80 Referrer Policy: no-referrer-when-downgrade Response Headers Content-Length: 3302 # 響應報文總長度 Content-Type: text/plain; charset=utf-8 # 要求文本顯示 字符串是UTF-8 Date: Sun, 08 Sep 2019 12:34:55 GMT Server: WSGIServer/0.2 CPython/3.6.4 #暴露服務器端信息 Request Headers Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3 # 客戶端瀏覽器可接受的類型和參數 Accept-Encoding: gzip, deflate # 可接受壓縮編碼 Accept-Language: zh-CN,zh;q=0.9,en;q=0.8 Cache-Control: max-age=0 Connection: keep-alive Cookie: csrftoken=Er5XLdEG211nWzgtJL1GFoxBgxFnnHbff2W7IiprrwTQbAAOzWWoHzihDrIxiK17 Host: 192.168.1.200 Upgrade-Insecure-Requests: 1 User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36 # 本身的user_agent
修改以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server ip='192.168.1.200' port=80 def app(environ,start_response): html='<h1>Hello World</h1>'.encode() start_response("200 OK", [('Content-Type','text/html; charset=utf-8')]) return [html] server=make_server(ip,port,app) # 實例化一個websever server.serve_forever() # 啓動 server.server_close() # 關閉 server.shutdown() # 刪除
結果以下
環境變量數據不少,都是存儲在字典中的,字典存取沒有對象的屬性使用方便,使用第三方webob,能夠把環境數據的解析,封裝成對象
pip install webob
將環境參數解析並封裝成request對象
GET方法,發送的數據是URL中的request handler中
request.get 就是一個字典MultiDict,裏面就封裝着查詢字符串POST 方法,"提交"的數據是放在request body裏面的,可是也同時可使用Query String
request.POST能夠獲取request Body中的數據,也是個字典MultiDict不關心什麼方法提交,只關心數據,可使用request.params,它裏面是全部提交數據的封裝
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response ip='192.168.1.200' port=80 def app(environ,start_response): request=Request(environ) print ("params:",request.params) #獲取傳輸的數據,query string 或者 POST 的body print ("method:",request.method) # 獲取請求方法 print ("path:",request.path) #獲取請求路徑 print ("user_agent:",request.user_agent) #獲取客戶端信息 print ("get data:",request.GET) #獲取get數據 print ("post data:",request.POST) # 獲取post body數據 html='<h1>Hello World</h1>'.encode() start_response("200 OK", [('Content-Type','text/html; charset=utf-8')]) return [html] server=make_server(ip,port,app) # 實例化一個websever server.serve_forever() # 啓動 server.server_close() # 關閉 server.shutdown() # 刪除
請求URL: http://192.168.1.200/admin/?username=mysql&password=123456
結果以下:
MultiDict 容許一個key存儲好幾個值
#!/usr/bin/poython3.6 #conding:utf-8 from webob.multidict import MultiDict md=MultiDict() md.add(1,'aaaa') md.add(1,'cccc') md.add(1,'bbbb') md.add(2,'aaaa') md.add(2,'bbbb') md.add(2,'cccc') md.add(3,'aaaa') for x in md.items(): print (x) print ('get:',md.get(1)) # 此處默認取最後一個加入的 print ('getall:',md.getall(1)) # 此處表示給據key取出全部 print (md.getone(3)) #只能有一個值,有多個值使用這個返回有問題
結果以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response ip='192.168.1.200' port=80 def app(environ,start_response): res=Response() start_response(res.status,res.headerlist) # 返回可迭代對象 html='<h1>Hello World</h1>'.encode("utf-8") return [html] server=make_server(ip,port,app) # 實例化一個websever server.serve_forever() # 啓動 server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response ip='192.168.1.200' port=80 def app(environ,start_response): res=Response('<h1>Hello World</h1>') # 寫法二 #res.body='<h1>Hello Python</h1>'.encode() #res.status_code=200 return res(environ,start_response) server=make_server(ip,port,app) # 實例化一個websever server.serve_forever() # 啓動 server.server_close() # 關閉 server.shutdown() # 刪除
結果以下
此裝飾器傳入一個request的參數,則返回一個Response 的返回值,實現了一進一出的狀況
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 @dec.wsgify def app(request:Request)->Response: return Response('<h1>hello python </h1>'.encode()) if __name__ == "__main__": server = make_server(ip, port, app) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
結果以下
什麼是路由,簡單的說,就是路怎麼走,就是按照不一樣的路徑分發數據
URL 就是不一樣資源的路徑,不一樣的路徑應該對應不一樣的應用程序來處理,因此代碼中須要增長對路徑的處理和分析
路徑 | 內容 |
---|---|
/ | 返回歡迎內容 |
/python | 返回hello python |
其餘路徑 | 返回404 |
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 @dec.wsgify def app(request:Request)->Response: res=Response() if request.path=="/": res.body='<h1>hello World</h1>'.encode() return res elif request.path=="/python": res.body='<h1>hello Python</h1>'.encode() return res else: res.status_code=404 res.body='<h1>Not Found</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, app) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res def show(request:Request): res=Response() res.status_code = 404 res.body = '<h1>Not Found</h1>'.encode() return res @dec.wsgify def app(request:Request)->Response: if request.path=="/": return showdefault(request) elif request.path=="/python": return showpython(request) else: return show(request) if __name__ == "__main__": server = make_server(ip, port, app) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res def show(request:Request): res=Response() res.status_code = 404 res.body = '<h1>Not Found</h1>'.encode() return res ROUTABLE={ '/' :showdefault, '/python' :showpython } @dec.wsgify def app(request:Request)->Response: return ROUTABLE.get(request.path,show)(request) if __name__ == "__main__": server = make_server(ip, port, app) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res def show(request:Request): res=Response() res.status_code = 404 res.body = '<h1>Not Found</h1>'.encode() return res ROUTABLE={} def register(path,fn): ROUTABLE[path]=fn register('/',showdefault) register('/python',showpython) @dec.wsgify def app(request:Request)->Response: return ROUTABLE.get(request.path,show)(request) if __name__ == "__main__": server = make_server(ip, port, app) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
思想: 將須要用戶本身編寫的東西放置在類的外邊,其餘的相關事件放置在類中
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec ip='192.168.1.200' port=80 class Application: ROUTABLE={} def show(self,request:Request): res=Response() res.status_code = 404 res.body = '<h1>Not Found</h1>'.encode() return res @classmethod def register(cls,path,fn): cls.ROUTABLE[path]=fn @dec.wsgify def __call__(self,request: Request) -> Response: return self.ROUTABLE.get(request.path,self.show)(request) def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res Application.register('/',showdefault) Application.register('/python',showpython) if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc ip='192.168.1.200' port=80 class Application: ROUTABLE={} @classmethod def register(cls,path,fn): cls.ROUTABLE[path]=fn @dec.wsgify def __call__(self,request: Request) -> Response: try: return self.ROUTABLE[request.path](request) except: raise exc.HTTPNotFound('訪問的資源不存在') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res Application.register('/',showdefault) Application.register('/python',showpython) if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc ip='192.168.1.200' port=80 class Application: ROUTABLE={} @classmethod def register(cls,path): def _register(handle): cls.ROUTABLE[path]=handle return handle return _register @dec.wsgify def __call__(self,request: Request) -> Response: try: return self.ROUTABLE[request.path](request) except: raise exc.HTTPNotFound('訪問的資源不存在') @Application.register('/') def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @Application.register('/python') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
到目前爲止,一個框架的雛形基本完成了
application是WSGI中的應用程序。可是這個應用程序已經變成了一個路由程序,處理邏輯已移動到了應用程序外了,而這部分就是留給程序員的部分。
目前實現的路由匹配,路徑匹配很是死板,使用正則表達式改造。導入re模塊,註冊時,存入的再也不是路徑字符串,而是pattern。
__call__方法中實現模式和傳入路徑的比較
compile 方法,編譯正則表達式
match 方法,必須從頭開始匹配, 只匹配一次
search方法,只匹配一次
fullmath 方法,要徹底匹配
findall方法,從頭開始找,找到全部匹配
字典的問題
若是使用字典,key若是是路徑,不能保證其順序,由於大多的匹配都是從嚴到寬,若是沒有必定的順序,則會致使問題正則表達式的預編譯問題
第一次使用會影響到用戶體驗,因此仍是要在註冊的時候編譯的。綜上,改用列表,元素使用二元祖(編譯後的正則對象,handler)
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,path): def _register(handle): cls.ROUTABLE.append((re.compile(path),handle)) return handle return _register @dec.wsgify def __call__(self,request: Request) -> Response: for pattern,hande in self.ROUTABLE: # 此處須要遍歷 matcher=pattern.match(request.path) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 return hande(request) raise exc.HTTPNotFound('訪問資源不存在') @Application.register('^/$') def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @Application.register('^/python$') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
請求方法,通常來講,既是是同一個URL,由於請求方法不一樣,處理方式也是不一樣的
假設一個URL。GET方法但願返回網頁內容,POST方法表示瀏覽器提交數據過來須要處理並存儲進數據庫,最終返回給客戶端存儲成功或者失敗信息,
換句話說,須要根據請求方法和正則同時匹配才能決定執行什麼樣的處理函數
方法 | 含義 |
---|---|
GET | 請求指定的頁面信息,並返回報頭和正文 |
HEAD | 相似於get請求,只不過返回的響應中沒有具體的內容,用於獲取報頭 |
POST | 向指定資源提交數據進行處理請求(例如提交表單或者上傳文件),數據被包含在請求正文中,POST請求可能會致使新的資源創建或者已有的資源的修改 |
PUT | 從客戶端向服務器端傳遞的數據取代指定的文檔的內容 |
DELETE | 請求服務器刪除指定的內容 |
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,path,method): # 此處加入請求方法 def _register(handle): cls.ROUTABLE.append((re.compile(path),handle,method)) return handle return _register @classmethod # 經過構造方法來完成對函數的註冊 def get(cls,path): return cls.register(path,'GET') @classmethod def post(cls,path): return cls.register(path,'POST') @classmethod def head(cls,path): return cls.register(path,'HEAD') @dec.wsgify def __call__(self,request: Request) -> Response: for pattern,hande,method in self.ROUTABLE: # 此處須要遍歷 if request.method==method.upper(): # 若是請求方法和對應註冊方法一致,則執行對應函數。 matcher=pattern.match(request.path) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 return hande(request) raise exc.HTTPNotFound('訪問資源不存在') @Application.get('^/$') def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @Application.get('^/python$') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @Application.post('^/python$') def showpython(request:Request): res=Response() res.body = '<h1>hello Python POST </h1>'.encode() return res @Application.post('^/') def showdefault(request:Request): res=Response() res.body = '<h1>hello World POST </h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
一個URL 能夠設定多種請求方法
要求:
1 若是什麼方法都不寫,至關於全部方法都支持
2 若是一個處理函數handler須要關聯多個請求方法method,以下:
@Application.register('^/$',('GET','PUT','DELETE')) @Application.register('^/python$',('GET','PUT','DELETE')) @Application.register('^/$',())
思路:
將method變爲methods,將位置參數變成可變參數便可代碼以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,path,*methods): # 此處加入請求方法 def _register(handle): cls.ROUTABLE.append((re.compile(path),handle,methods)) return handle return _register @classmethod # 經過構造方法來完成對函數的註冊 def get(cls,path): return cls.register(path,'GET','POST') @classmethod def post(cls,path): return cls.register(path,'POST') @classmethod def head(cls,path): return cls.register(path,'HEAD') @dec.wsgify def __call__(self,request: Request) -> Response: for pattern,hande,methods in self.ROUTABLE: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 return hande(request) raise exc.HTTPNotFound('訪問資源不存在') @Application.get('^/$') def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @Application.get('^/python') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
結果以下:
支持正則表達式的捕獲,
在框架回調__call__時,拿到request.path和正則的模式匹配後,就能夠提取分組了如何處理分組?
應用程序就是handler對應的不一樣的函數,其參數request是同樣的,將捕獲的數據動態增長到request對象中便可。用動態增長屬性,爲request增長args,kwargs屬性,在handler中使用的時候,就能夠直接熊屬性中,將args,kwargs拿出來就能夠直接使用了
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,path,*methods): # 此處加入請求方法 def _register(handle): cls.ROUTABLE.append((re.compile(path),handle,methods)) return handle return _register @classmethod # 經過構造方法來完成對函數的註冊 def get(cls,path): return cls.register(path,'GET','POST') @classmethod def post(cls,path): return cls.register(path,'POST') @classmethod def head(cls,path): return cls.register(path,'HEAD') @dec.wsgify def __call__(self,request: Request) -> Response: for pattern,hande,methods in self.ROUTABLE: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() # 此處獲取元祖元素 request.kwargs=matcher.groupdict() # 此處獲取字典元素進行處理 return hande(request) raise exc.HTTPNotFound('訪問資源不存在') @Application.get('^/$') def showdefault(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @Application.get('^/python') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
所謂的路由分組,就是按照前綴分別映射
需求
URL 爲 /product/123456
須要將產品ID提取出來
分析
這個URL能夠看作是一級分組路由,生產環境中可使用了
如
product=Router('/product') #匹配前綴 /product
product.get('/(?P<id>\d+)') # 匹配路徑爲/product/123456
常見的一級目錄
/admin #後臺管理
/product 產品
這些目錄都是/跟目錄的下一級目錄,暫時稱爲前綴prefix
如何創建prefix和URL 之間的隸屬關係
一個prefix下能夠有若干個URL。這些URL都是屬於這個prefix中的
創建一個Router類,裏面保存Prefix,同時保存URL和handler的關係之前。註冊的方法都是application的類方法,也就是全部映射信息都保存在一個類屬性中ROUTABLE中,可是如今要爲不一樣的前綴就是不一樣的實例,所以全部註冊方法,都是實例的方法,路由包實例本身管理
application 中如今只須要保存全部註冊的Router對象就好了,__call__方法依然是回調入口,在其中遍歷全部的Router,找到路徑匹配的Router實例,讓Router實例返回Response 對象便可
代碼以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 print (self.__prefix) @property def prefix(self): return self.__prefix def register(self,path,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): self.__routertable.append((re.compile(path),handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for pattern,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) print ('prefix',self.prefix) print (request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 print(matcher) request.args=matcher.group() request.kwargs=matcher.groupdict() return hande(request) class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,router:Router): cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: return response raise exc.HTTPNotFound('訪問資源不存在') # 註冊前綴 index=Router('/') pyth=Router('/python') admin=Router('/admin') #將前綴加入對應列表中 Application.register(pyth) Application.register(admin) Application.register(index) # 寫handler @index.get('/(\w+)') def showpython(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/(\d+)') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/(\d+)') def showadmin(request:Request): res=Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
經過此類,可以使得kwargs這個字典,不使用[]訪問元素,使用.號訪問元素,如同屬性同樣訪問
#!/usr/bin/poython3.6 #conding:utf-8 class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常,此處的調用是兩個步驟,第一個是self._dict 而後會調用各類方法最終造成死循環,若是專用的字典中有的話,則其中不會訪問 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented d={ 'a':1, 'b':2, 'c':3 } x=DictOrd(d) print (x.__dict__) print (DictOrd.__dict__) print (x.a) print (x.b)
結果以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 print (self.__prefix) @property def prefix(self): return self.__prefix def register(self,path,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): self.__routertable.append((re.compile(path),handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for pattern,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) print ('prefix',self.prefix) print (request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 print(matcher) request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) # 此處經過修改後的字典,使得下面的訪問能夠直接使用.來進行訪問而不是使用[key]的方式進行相關的訪問操做 return hande(request) class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,router:Router): cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: return response raise exc.HTTPNotFound('訪問資源不存在') # 註冊前綴 #將前綴加入對應列表中 index=Router('/') pyth=Router('/python') admin=Router('/admin') Application.register(pyth) Application.register(admin) Application.register(index) @index.get('/(\w+)') def showpython(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/(\d+)') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/(\d+)') def showadmin(request:Request): res=Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
問題
目前路由匹配使用正則表達式定義,不友好,不少用戶不會使用正則表達式,可否簡化分析
生產環境中。URL是規範的,不能隨意書寫,路徑是有意義的,尤爲是對restful風格,因此,要對URL規範
如 product/111102243454343 ,這就是一種規範,要求第一段是業務,第二段是ID。設計
路徑規範化,以下定義
/student/{name:str}/{id:int}
類型設計。支持str,word,int,float,any類型
經過這種定義,可讓用戶定義簡化了,也規範了,背後的轉換是編程者實現的
類型 | 含義 | 對應正則 |
---|---|---|
str | 不包含/的任意字符 | [^/]+ |
word | 字母和數字 | \w+ |
int | 純數字,正負數 | [+-]?\d+ |
float | 正負號,數字,包含. | [+-]?\d+.\d+ |
any | 包含/的任意字符 | .+ |
保存類型
類型 | 對應類型 |
---|---|
str | str |
word | str |
int | int |
float | float |
any | str |
#!/usr/local/bin/python3.6 #coding:utf-8 import re s='/student/{name:abcded}/xxxx/{id:12345}' s1='/student/xxxx/{id:12345}' s2='/student/xxxx/12344' s3='/student/{name:aaa}/xxxx/{id:1245}' TYPEPATTERNS= { 'str' :r'[^/]+', 'word' :r'\w+', 'int' :r'[+-]?\d+', 'float' : r'[+-]?\d+.\d+', 'any' : r'.+' } TYPECAST= { 'str' :str, 'word': str, 'int' :int, 'float' :float, 'any' :str } pattern=re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(kv:str): name,_,type=kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name,TYPEPATTERNS.get(type,'\w+')),name,TYPECAST.get(type,str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元組 def parse(src:str): start=0 res='' translator= {} while True: matcher=pattern.search(src,start) # start表示偏移量 if matcher: res+=matcher.string[start:matcher.start()] #對匹配到的字符串進行切割處理 tmp=transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res+=tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]]=tmp[2] # 此處保存的是名稱和類型的字典 start=matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res,translator else: # 若不存在,也返回 return res,translator print (parse(s)) print (parse(s1)) print (parse(s2)) print (parse(s3))
結果以下
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) return hande(request) class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 @classmethod def register(cls,router:Router): cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: return response raise exc.HTTPNotFound('訪問資源不存在') # 註冊前綴 #將前綴加入對應列表中 index=Router('/') pyth=Router('/python') admin=Router('/admin') Application.register(pyth) Application.register(admin) Application.register(index) @index.get('/\w+') def showpython(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(request:Request): res=Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
處理流程
客戶端發起請求,被容器調度給Application的call
Application 中便利全部註冊的router,router經過match來判斷是不是本身處理的,經過查看request的請求來匹配註冊前綴,若符合條件,則匹配對應的請求方法,若方法符合,則匹配對應的URL二級目錄,並返回對應的函數,handler處理後,返回response,applicationde拿着這個response數據,返回給原始的wsgi。
攔截器,就是要在請求處理環節的某處加入處理,有多是中斷手續的處理
根據攔截點不一樣,分爲:
1 請求攔截
2 響應攔截
根據影響面分爲:
1 全局攔截
在application中攔截
2 局部攔截
在Router中攔截
相關圖形
前面的是application層面的攔截,及全局攔截,。後面是TOUTER層面的攔截,及局部攔截,
攔截器能夠是多個,多個攔截器是順序的
數據response前執行的的命名爲preinterceptor ,以後的命名爲postinterceptor。
1 application 和Router 類直接加入
把攔截器的相關方法,屬性分別調價到相關的類中2 Mixin
Application 和Router類都須要這個攔截器功能,這兩個類沒什麼關係,可使用Mixin方式,將屬性,方法組合起來
可是,application類攔截器適合使用第二種方式,DNA是Router的攔截器每一個實例都是不一樣的,因此使用第一種方式實現
當出現多繼承時,Mixin中MRO規則會直接使用第一個,而忽略其餘的__init__方法。
攔截器的函數是相對獨立的,其至關因而相對透明的,用一個的輸出和N的輸出都應該可以和handler進行處理
引入app,是爲了之後從application上獲取一些全局信息,其application的實例資源。
來的輸入和輸出都是request
def fn(app,request:Request)->Request: pass
去的輸入和輸出都是response
def fn(app,request:Request,response:Response)-> Response: pass
爲了把一些應數據,配置數據,數據庫鏈接提供給全局共享數據提供全部對象使用,增長一個字典,存儲共享數據。將環境變量傳遞下去。
爲了方便訪問,提供字典的屬性化訪問的類,由於這個字典是可寫的,和前面的類不同。
application最多的應該作的是單實例模式,及就是一個實例的處理模式,若果是要用多實例,則須要使用信號量或其餘進行處理
class Context(dict): #繼承內部類,使得類可以提供一種能力可以直接的屬性訪問方式,讀取配置文件的能力 def __getattr__(self, item): # 經過.的方式進行訪問 try: return self[item] # 本身的字典訪問方式給請求端 except KeyError: # 屬性訪問的方式致使的問題 raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value #處理修改和添加問題
Router沒一個實例中增長上下文屬性,實例本身使用
可是Router實例如何使用全局上下文
使用新的處理方法,每個Router實例的上下文字典內部關聯一個全局字典的引用,若是本身的字典找不到,就去全局尋找
那Router實例何時關聯全局字典比較合適
在路由註冊的時候便可基本代碼以下
class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item]
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Context(dict): # 用於存儲共享數據,app使用 def __getattr__(self, item): try: return self[item] except KeyError: raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value ####################上述兩種字典的不一樣實現方式處理########################### class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個。Router實例使用 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item] class Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 self.ctx=NestedContext() # 未綁定全局的上下文,在註冊的時候進行處理 #實例本身使用的攔截器。在match處進行攔截 self.preinterceptor=[] self.postinterceptor=[] # 裝飾器須要有返回值 def reg_preinterceptor(self, fn): # fn前半段兩個參數,後半段三個參數,裝飾器須要返回值 self.preinterceptor.append(fn) return fn def reg_postinterceptor(self, fn): self.postinterceptor.append(fn) return fn # TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for fn in self.preinterceptor: # 攔截器處理 request=fn(self.ctx,request) for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) response=hande(self.ctx,request) #優先使用本身的屬性 for fn in self.postinterceptor: response=fn(self.ctx,request,response) return response class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 ctx=Context() #實例的攔截器 PREINTERCEPTOR=[] POSTINTERCEPTOR=[] # 攔截器的註冊 @classmethod def reg_preinterceptor(cls,fn): # fn前半段兩個參數,後半段三個參數 cls.PREINTERCEPTOR.append(fn) return fn @classmethod def reg_postinterceptor(cls,fn): cls.POSTINTERCEPTOR.append(fn) return fn # 函數須要返回,其自己並無變更 def __init__(self,**kwargs): self.ctx.app=self for k,v in kwargs.items(): self.ctx[k]=v #添加註冊功能 @classmethod def register(cls,router:Router): router.ctx.relate(cls.ctx) #將上述的CTX添加進來,用於屬性的訪問控制及上述的NestedContext,將全局的上下文綁定給每個router實例 # 其在router本身初始化時就本身建立 router.ctx.router=router #在本身的字典中中引用本身 cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for fn in self.PREINTERCEPTOR: # 註冊函數, request=fn(self.ctx,request) #第一個是全局的,第二個是本身的,定義的,須要request不變透明化 #fn(self.ctx,request) 此處此種寫法容易引發別人的誤會 for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: #返回的函數進行處理 for fn in self.POSTINTERCEPTOR: # 此到處理response相關的方法 response=fn(self.ctx.request,response) return response raise exc.HTTPNotFound('訪問資源不存在') # 註冊前綴 #將前綴加入對應列表中 index=Router('/') pyth=Router('/python') admin=Router('/admin') Application.register(pyth) Application.register(admin) Application.register(index) #添加攔截器 @Application.reg_preinterceptor #全局起始攔截器 def showhandler(ctx:Context,request:Request)-> Request: print (request.path) print (request.user_agent) return request # 返回爲request,只有request @pyth.reg_preinterceptor # Router 層面的攔截器 def showprefix(ctx:NestedContext,request:Request)->Request: print ('~~~~~~~~~~~~prefix = {}'.format(ctx.router.prefix)) # 此處是打印本身的前綴 return request @index.get('/\w+') def showpython(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(request:Request): res=Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
做爲一個框架,更多的功能應該是從外部加入
1 不可能些的很是完善
2 非必要的都應該動態加入
因此,提供一個擴展接口很是重要
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re ip='192.168.1.200' port=80 class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Context(dict): # 用於存儲共享數據,app使用 def __getattr__(self, item): try: return self[item] except KeyError: raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value ####################上述兩種字典的不一樣實現方式處理########################### class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個。Router實例使用 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item] class Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 self.ctx=NestedContext() # 未綁定全局的上下文,在註冊的時候進行處理 #實例本身使用的攔截器。在match處進行攔截 self.preinterceptor=[] self.postinterceptor=[] # 裝飾器須要有返回值 def reg_preinterceptor(self, fn): # fn前半段兩個參數,後半段三個參數,裝飾器須要返回值 self.preinterceptor.append(fn) return fn def reg_postinterceptor(self, fn): self.postinterceptor.append(fn) return fn # TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for fn in self.preinterceptor: # 攔截器處理 request=fn(self.ctx,request) for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) response=hande(self.ctx,request) #優先使用本身的屬性 for fn in self.postinterceptor: response=fn(self.ctx,request,response) return response class Application: ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 ctx=Context() #實例的攔截器 PREINTERCEPTOR=[] POSTINTERCEPTOR=[] @classmethod # 增長擴展功能模塊,經過名字的方式加載進來 def extend(cls,name,ext): cls.ctx[name]=ext # 攔截器的註冊 @classmethod def reg_preinterceptor(cls,fn): # fn前半段兩個參數,後半段三個參數 cls.PREINTERCEPTOR.append(fn) return fn @classmethod def reg_postinterceptor(cls,fn): cls.POSTINTERCEPTOR.append(fn) return fn # 函數須要返回,其自己並無變更 def __init__(self,**kwargs): self.ctx.app=self for k,v in kwargs.items(): self.ctx[k]=v #添加註冊功能 @classmethod def register(cls,router:Router): router.ctx.relate(cls.ctx) #將上述的CTX添加進來,用於屬性的訪問控制及上述的NestedContext,將全局的上下文綁定給每個router實例 # 其在router本身初始化時就本身建立 router.ctx.router=router #在本身的字典中中引用本身 cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for fn in self.PREINTERCEPTOR: # 註冊函數, request=fn(self.ctx,request) #第一個是全局的,第二個是本身的,定義的,須要request不變透明化 #fn(self.ctx,request) 此處此種寫法容易引發別人的誤會 for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: #返回的函數進行處理 for fn in self.POSTINTERCEPTOR: # 此到處理response相關的方法 response=fn(self.ctx.request,response) return response raise exc.HTTPNotFound('訪問資源不存在') # 註冊前綴 #將前綴加入對應列表中 index=Router('/') pyth=Router('/python') admin=Router('/admin') Application.register(pyth) Application.register(admin) Application.register(index) #添加攔截器 @Application.reg_preinterceptor #全局起始攔截器 def showhandler(ctx:Context,request:Request)-> Request: print (request.path) print (request.user_agent) return request # 返回爲request,只有request @pyth.reg_preinterceptor # Router 層面的攔截器 def showprefix(ctx:NestedContext,request:Request)->Request: print ('~~~~~~~~~~~~prefix = {}'.format(ctx.router.prefix)) # 此處是打印本身的前綴 return request @index.get('/\w+') def showpython(request:Request): res=Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(request:Request): res=Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(request:Request): res=Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": server = make_server(ip, port, Application()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
在pycharm中建立一個包,包名爲testweb
在init.py文件中,修改Application爲TestWeb
經過此種方式暴露類
class TestWeb: # 類屬性方法把類暴露出去 Router=_Router Request=Request Response=Response NestedContext=NestedContext Context=Context
以供別人調用
外層新建app,將須要調用的都建立在app中實現,及就是使用此模塊的人
目錄
app.py 中實現的代碼
from wsgiref.simple_server import make_server from testweb import TestWeb # 註冊前綴 #將前綴加入對應列表中 index=TestWeb.Router('/') pyth=TestWeb.Router('/python') admin=TestWeb.Router('/admin') TestWeb.register(pyth) TestWeb.register(admin) TestWeb.register(index) #添加攔截器 @TestWeb.reg_preinterceptor #全局起始攔截器 def showhandler(ctx:TestWeb.Context,request:TestWeb.Request)-> TestWeb.Request: print (request.path) print (request.user_agent) return request # 返回爲request,只有request @pyth.reg_preinterceptor # Router 層面的攔截器 def showprefix(ctx:TestWeb.NestedContext,request:TestWeb.Request)->TestWeb.Request: print ('~~~~~~~~~~~~prefix = {}'.format(ctx.router.prefix)) # 此處是打印本身的前綴 return request @index.get('/\w+') def showpython(request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": ip = '192.168.1.200' port = 80 server = make_server(ip, port, TestWeb()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
testweb中_init_.py中的內容
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Context(dict): # 用於存儲共享數據,app使用 def __getattr__(self, item): try: return self[item] except KeyError: raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value ####################上述兩種字典的不一樣實現方式處理########################### class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個。Router實例使用 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item] class _Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 self.ctx=NestedContext() # 未綁定全局的上下文,在註冊的時候進行處理 #實例本身使用的攔截器。在match處進行攔截 self.preinterceptor=[] self.postinterceptor=[] # 裝飾器須要有返回值 def reg_preinterceptor(self, fn): # fn前半段兩個參數,後半段三個參數,裝飾器須要返回值 self.preinterceptor.append(fn) return fn def reg_postinterceptor(self, fn): self.postinterceptor.append(fn) return fn # TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for fn in self.preinterceptor: # 攔截器處理 request=fn(self.ctx,request) for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) response=hande(self.ctx,request) #優先使用本身的屬性 for fn in self.postinterceptor: response=fn(self.ctx,request,response) return response class TestWeb: # 類屬性方法把類暴露出去 Router=_Router Request=Request Response=Response NestedContext=NestedContext Context=Context ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 ctx=Context() #實例的攔截器 PREINTERCEPTOR=[] POSTINTERCEPTOR=[] @classmethod # 增長擴展功能模塊 def extend(cls,name,ext): cls.ctx[name]=ext # 攔截器的註冊 @classmethod def reg_preinterceptor(cls,fn): # fn前半段兩個參數,後半段三個參數 cls.PREINTERCEPTOR.append(fn) return fn @classmethod def reg_postinterceptor(cls,fn): cls.POSTINTERCEPTOR.append(fn) return fn # 函數須要返回,其自己並無變更 def __init__(self,**kwargs): self.ctx.app=self for k,v in kwargs.items(): self.ctx[k]=v #添加註冊功能 @classmethod def register(cls,router:Router): router.ctx.relate(cls.ctx) #將上述的CTX添加進來,用於屬性的訪問控制及上述的NestedContext,將全局的上下文綁定給每個router實例 # 其在router本身初始化時就本身建立 router.ctx.router=router #在本身的字典中中引用本身 cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for fn in self.PREINTERCEPTOR: # 註冊函數, request=fn(self.ctx,request) #第一個是全局的,第二個是本身的,定義的,須要request不變透明化 #fn(self.ctx,request) 此處此種寫法容易引發別人的誤會 for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: #返回的函數進行處理 for fn in self.POSTINTERCEPTOR: # 此到處理response相關的方法 response=fn(self.ctx.request,response) return response raise exc.HTTPNotFound('訪問資源不存在')
此處屬於模塊的附加功能
import json def jsonify(**kwargs): content=json.dumps(kwargs) response=Response() response.content_type="application/json" # 規定返回結果 response.charset='utf-8' response.body="{}".format(content).encode() # 此處不能添加,添加了就不是json格式的數據了 return Response()
_init_.py中的配置
#!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Context(dict): # 用於存儲共享數據,app使用 def __getattr__(self, item): try: return self[item] except KeyError: raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value ####################上述兩種字典的不一樣實現方式處理########################### class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個。Router實例使用 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item] class _Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 self.ctx=NestedContext() # 未綁定全局的上下文,在註冊的時候進行處理 #實例本身使用的攔截器。在match處進行攔截 self.preinterceptor=[] self.postinterceptor=[] # 裝飾器須要有返回值 def reg_preinterceptor(self, fn): # fn前半段兩個參數,後半段三個參數,裝飾器須要返回值 self.preinterceptor.append(fn) return fn def reg_postinterceptor(self, fn): self.postinterceptor.append(fn) return fn # TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for fn in self.preinterceptor: # 攔截器處理 request=fn(self.ctx,request) for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) response=hande(self.ctx,request) #優先使用本身的屬性 for fn in self.postinterceptor: response=fn(self.ctx,request,response) return response import json def jsonify(**kwargs): content=json.dumps(kwargs) response=Response() response.content_type="application/json" # 規定返回結果 response.charset='utf-8' response.body="{}".format(content).encode() # 此處不能添加,添加了就不是json格式的數據了 return Response() class TestWeb: # 類屬性方法把類暴露出去 Router=_Router Request=Request Response=Response NestedContext=NestedContext Context=Context jsonify=jsonify ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 ctx=Context() #實例的攔截器 PREINTERCEPTOR=[] POSTINTERCEPTOR=[] @classmethod # 增長擴展功能模塊 def extend(cls,name,ext): cls.ctx[name]=ext # 攔截器的註冊 @classmethod def reg_preinterceptor(cls,fn): # fn前半段兩個參數,後半段三個參數 cls.PREINTERCEPTOR.append(fn) return fn @classmethod def reg_postinterceptor(cls,fn): cls.POSTINTERCEPTOR.append(fn) return fn # 函數須要返回,其自己並無變更 def __init__(self,**kwargs): self.ctx.app=self for k,v in kwargs.items(): self.ctx[k]=v #添加註冊功能 @classmethod def register(cls,router:Router): router.ctx.relate(cls.ctx) #將上述的CTX添加進來,用於屬性的訪問控制及上述的NestedContext,將全局的上下文綁定給每個router實例 # 其在router本身初始化時就本身建立 router.ctx.router=router #在本身的字典中中引用本身 cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for fn in self.PREINTERCEPTOR: # 註冊函數, request=fn(self.ctx,request) #第一個是全局的,第二個是本身的,定義的,須要request不變透明化 #fn(self.ctx,request) 此處此種寫法容易引發別人的誤會 for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: #返回的函數進行處理 for fn in self.POSTINTERCEPTOR: # 此到處理response相關的方法 response=fn(self.ctx.request,response) return response raise exc.HTTPNotFound('訪問資源不存在')
app.py中的值
from wsgiref.simple_server import make_server from testweb import TestWeb # 註冊前綴 #將前綴加入對應列表中 index=TestWeb.Router('/') pyth=TestWeb.Router('/python') admin=TestWeb.Router('/admin') TestWeb.register(pyth) TestWeb.register(admin) TestWeb.register(index) #添加攔截器 @TestWeb.reg_preinterceptor #全局起始攔截器 def showhandler(ctx:TestWeb.Context,request:TestWeb.Request)-> TestWeb.Request: print (request.path) print (request.user_agent) return request # 返回爲request,只有request @pyth.reg_preinterceptor # Router 層面的攔截器 def showprefix(ctx:TestWeb.NestedContext,request:TestWeb.Request)->TestWeb.Request: print ('~~~~~~~~~~~~prefix = {}'.format(ctx.router.prefix)) # 此處是打印本身的前綴 return request @admin.reg_postinterceptor # json的處理 def showjson(NestedContext,request,response): body=response.body.decode() # 此處返回的是一個字節,須要解碼 return TestWeb.jsonify(body=body) # 此處必須傳入一個字典。不然會出問題,body是鍵,文本內容是值 @index.get('/\w+') def showpython(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": ip = '192.168.1.200' port = 80 server = make_server(ip, port, TestWeb()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
1 熟悉WSGI的編程接口
2 強化模塊化,類封裝思想
3 增長分析業務的能力這個框架基本劇本了WSGI WEB 框架的基本功能,其餘框架都相似。
權限驗證,SQL注入檢測的功能使用攔截器過濾。
在 testweb包外建立setup.py 在 testweb包內建立web文件
結構以下
web文件內容以下
#!/usr/bin/poython3.6 #conding:utf-8 #!/usr/bin/poython3.6 #conding:utf-8 from wsgiref.simple_server import make_server from webob import Request,Response,dec,exc import re class DictObj: def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去,可能會有衝突致使屬性覆蓋的問題 if not isinstance(d,dict): self.__dict__['_dict']={} # 此處不能是雙下劃綫,設置類屬性字典 else: self.__dict__['_dict']=d #將字典加入到實例屬性列表中 def __getattr__(self, item): #此處是經過點號訪問的 try: return self._dict[item] # 經過d.x訪問,若存在,則直接返回,若不存在,則拋出異常 except KeyError: #當其鍵不存在的時候 raise AttributeError('Attribute {} Not Found'.format(item)) def __setattr__(self, key, value): #此處是點號修改的 # 不容許設置屬性,set表示未實現 raise NotImplemented class Context(dict): # 用於存儲共享數據,app使用 def __getattr__(self, item): try: return self[item] except KeyError: raise ArithmeticError('Attribe {} Not Found'.format(item)) def __setattr__(self, key, value): self[key]=value ####################上述兩種字典的不一樣實現方式處理########################### class NestedContext(Context): #繼承上述屬性,什麼邏輯不同就覆蓋那個。Router實例使用 def __init__(self,globalcontext:Context=None): super().__init__() self.relate(globalcontext) def relate(self,globalcontext:Context=None): self.globalcontext=globalcontext def __getattr__(self, item): if item in self.keys(): return self[item] return self.globalcontext[item] class _Router: def __init__(self,prefix:str): self.__prefix=prefix.rstrip('/\\') # 去除prefix及一級目錄後面的\\和多餘的/ self.__routertable=[] #此處用於保存handler,pattern,method的信息 self.ctx=NestedContext() # 未綁定全局的上下文,在註冊的時候進行處理 #實例本身使用的攔截器。在match處進行攔截 self.preinterceptor=[] self.postinterceptor=[] # 裝飾器須要有返回值 def reg_preinterceptor(self, fn): # fn前半段兩個參數,後半段三個參數,裝飾器須要返回值 self.preinterceptor.append(fn) return fn def reg_postinterceptor(self, fn): self.postinterceptor.append(fn) return fn # TYPEPATTERNS = { 'str': r'[^/]+', 'word': r'\w+', 'int': r'[+-]?\d+', 'float': r'[+-]?\d+.\d+', 'any': r'.+' } TYPECAST = { 'str': str, 'word': str, 'int': int, 'float': float, 'any': str } pattern = re.compile('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關用戶信息的狀況,此處匹配到的只是一級目錄的相關信息 def transfrom(self,kv: str): name, _, type = kv.strip('/{}').partition(':') # 此處用於替換操作,此處返回一個列表,經過參數解構來收集,後面是找到第一個後進行分割操作 return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type, str) # 此處的format只是構造name和對應正則匹配的字典,此處返回的是一個三元 def parse(self,src: str): start = 0 res = '' translator = {} while True: matcher = self.pattern.search(src, start) # start表示偏移量 if matcher: res += matcher.string[start:matcher.start()] # 對匹配到的字符串進行切割處理 tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結果的集合,此出返回一個三元組 res += tmp[0] # 此處保存的是名稱和正則的元組 translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典 start = matcher.end() # 此處再次匹配,則須要進行初始化繼續匹配的操作 else: # 若不能匹配,則返回 break if res: # 若存在,則返回 return res, translator # res中保存URL,translator中保存名稱和類型的對應關係 else: # 若不存在,也返回 return res, translator @property def prefix(self): return self.__prefix def register(self,rule,*methods): # 此處用於註冊二級目錄對應的值 def _register(handle): pattern,translator=self.parse(rule) #此處經過對應的規則來處理相關配置,pattern中包含的是實際的URL路徑,translator 中包含分組名稱和對應類型的匹配 self.__routertable.append((re.compile(pattern),translator,handle,methods)) return handle return _register def get(self,path): return self.register(path,'GET') def post(self,path): return self.register(path,'POST') def head(self,path): return self.register(path,'HEAD') def match(self,request:Request): if not request.path.startswith(self.__prefix): #判斷其是否URL一級目錄匹配註冊的prefix,若不匹配則返回爲None return for fn in self.preinterceptor: # 攔截器處理 request=fn(self.ctx,request) for pattern,translator,hande,methods in self.__routertable: # 此處須要遍歷 if not methods or request.method in methods: matcher=pattern.match(request.path.replace(self.prefix,"",1)) if matcher: # 此處若能匹配到,則爲True,則能夠進行下一步 request.args=matcher.group() print (type(matcher.groupdict())) request.kwargs=DictObj(matcher.groupdict()) newdict={} for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典,K是分組名稱,V是匹配的結果 newdict[k]=translator[k](v) #分組匹配結果,經過分組的名稱獲取對應的類型進行對其值進行操做並保存 request.vars=DictObj(newdict) response=hande(self.ctx,request) #優先使用本身的屬性 for fn in self.postinterceptor: response=fn(self.ctx,request,response) return response class TestWeb: # 類屬性方法把類暴露出去 Router=_Router Request=Request Response=Response NestedContext=NestedContext Context=Context ROUTABLE=[] # 此處修改爲列表的形式比較適合順序匹配 ctx=Context() #實例的攔截器 PREINTERCEPTOR=[] POSTINTERCEPTOR=[] @classmethod # 增長擴展功能模塊 def extend(cls,name,ext): cls.ctx[name]=ext # 攔截器的註冊 @classmethod def reg_preinterceptor(cls,fn): # fn前半段兩個參數,後半段三個參數 cls.PREINTERCEPTOR.append(fn) return fn @classmethod def reg_postinterceptor(cls,fn): cls.POSTINTERCEPTOR.append(fn) return fn # 函數須要返回,其自己並無變更 def __init__(self,**kwargs): self.ctx.app=self for k,v in kwargs.items(): self.ctx[k]=v #添加註冊功能 @classmethod def register(cls,router:Router): router.ctx.relate(cls.ctx) #將上述的CTX添加進來,用於屬性的訪問控制及上述的NestedContext,將全局的上下文綁定給每個router實例 # 其在router本身初始化時就本身建立 router.ctx.router=router #在本身的字典中中引用本身 cls.ROUTABLE.append(router) # 此處用於調用上述的函數,完成數據的初始化並傳遞相關的參數prefix參數 @dec.wsgify def __call__(self,request: Request) -> Response: for fn in self.PREINTERCEPTOR: # 註冊函數, request=fn(self.ctx,request) #第一個是全局的,第二個是本身的,定義的,須要request不變透明化 #fn(self.ctx,request) 此處此種寫法容易引發別人的誤會 for router in self.ROUTABLE: # 遍歷router傳輸相關參數 response=router.match(request) # 此處返回爲handler的函數值 if response: #返回的函數進行處理 for fn in self.POSTINTERCEPTOR: # 此到處理response相關的方法 response=fn(self.ctx.request,response) return response raise exc.HTTPNotFound('訪問資源不存在') if __name__ == "__main__": pass
_init_.py文件
from .web import TestWeb # 此處外部訪問只能使用TestWeb進行各類處理,而能使用Request或Response import json def jsonify(**kwargs): content=json.dumps(kwargs) response=TestWeb.Response() response.content_type="application/json" # 規定返回結果 response.charset='utf-8' response.body="{}".format(content).encode() # 此處不能添加,添加了就不是json格式的數據了 return TestWeb.Response()
app文件內容
from wsgiref.simple_server import make_server from testweb import TestWeb,jsonify # 註冊前綴 #將前綴加入對應列表中 index=TestWeb.Router('/') pyth=TestWeb.Router('/python') admin=TestWeb.Router('/admin') TestWeb.register(pyth) TestWeb.register(admin) TestWeb.register(index) #添加攔截器 @TestWeb.reg_preinterceptor #全局起始攔截器 def showhandler(ctx:TestWeb.Context,request:TestWeb.Request)-> TestWeb.Request: print (request.path) print (request.user_agent) return request # 返回爲request,只有request @pyth.reg_preinterceptor # Router 層面的攔截器 def showprefix(ctx:TestWeb.NestedContext,request:TestWeb.Request)->TestWeb.Request: print ('~~~~~~~~~~~~prefix = {}'.format(ctx.router.prefix)) # 此處是打印本身的前綴 return request @admin.reg_postinterceptor # json的處理 def showjson(NestedContext,request,response): body=response.body.decode() # 此處返回的是一個字節,須要解碼 return jsonify(body=body) # 此處必須傳入一個字典。不然會出問題,body是鍵,文本內容是值 @index.get('/\w+') def showpython(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello World</h1>'.encode() return res @pyth.get('/\d+') def showpython(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello Python</h1>'.encode() return res @admin.get('/\d+') def showadmin(NestedContext,request:TestWeb.Request): res=TestWeb.Response() res.body = '<h1>hello admin</h1>'.encode() return res if __name__ == "__main__": ip = '192.168.1.200' port = 80 server = make_server(ip, port, TestWeb()) # 實例化一個websever try: server.serve_forever() # 啓動 except KeyboardInterrupt: pass finally: server.server_close() # 關閉 server.shutdown() # 刪除
setup.py 內容
#!/usr/bin/poython3.6 #conding:utf-8 from distutils.core import setup setup( name='testweb', # 名字 version='0.1.0', #版本 description='testweb', #打包列表 author='zhang', # 做者 author_email='12345678910@163.com', # # url 表示包幫助文檔路徑 packages=['testweb'] )
打包
python setup.py sdist
安裝
pip install dist/test
複製到另外一個環境安裝查看