溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
web 也叫CS開發(fā)
CS 及客戶端，服務(wù)器端編程
客戶端，服務(wù)器端之間需要socket，約定協(xié)議，版本(往往使用的協(xié)議是TCP或UDP)，指定地址和端口，就可以通信了客戶端，服務(wù)端傳輸數(shù)據(jù)，數(shù)據(jù)可以有一定的格式，雙方必須約定好
B=Browser,Browser是一種特殊的客戶端，支持HTTP(S)協(xié)議，能夠通過URL 向服務(wù)器端發(fā)起請求，等待服務(wù)端返回HTML等數(shù)據(jù)，并在瀏覽器內(nèi)可視化展示程序
S=server，Server支持HTTP(S)協(xié)議，能夠接受眾多客戶端發(fā)起的HTTP請求，經(jīng)過處理，將HTML等數(shù)據(jù)返回給瀏覽器
本質(zhì)上來說，BS是一種特殊的CS，及客戶端必須是一種支持HTTP協(xié)議且能夠解析并渲染HTML的軟件，服務(wù)端必須是能夠接受客戶端HTTP訪問的服務(wù)器軟件
HTTP 底層使用TCP傳輸，需要通過相應(yīng)的規(guī)則來處理
HTTP 超文本傳輸協(xié)議，在文本的基礎(chǔ)上做一些突破，相關(guān)的渲染處理，斷行等。使用標(biāo)簽的方式進(jìn)行規(guī)定的處理，文本已經(jīng)有了一個格式，瀏覽器中必須一種能力，支持這種文本的處理和渲染。
瀏覽器必須支持HTTP協(xié)議，必須能夠理解超文本并將其繪制出來
BS 開發(fā)分為兩端開發(fā)
1 客戶端開發(fā)，或稱為前端開發(fā)
2 服務(wù)端開發(fā)，python可以學(xué)習(xí)WSGI，Django，F(xiàn)lask，Tornado等python WEB 框架
WSGI， web Server Gateway interface，可以看做是一種底層協(xié)議，它規(guī)定了服務(wù)器程序和應(yīng)用程序各自實現(xiàn)什么借口，python稱為wsgirefflask: 基于WSGI ，微框架
Django：基于WSGI，開源的WEB框架
在http1.1之前，都是一個請求一個連接，而TCP的鏈接創(chuàng)建銷毀成本高，對服務(wù)器影響較大，因此自從http1.1開始，支持keep-alive，默認(rèn)也開啟，一個連接打開后，會保持一段時間，瀏覽器再訪問該服務(wù)器資源就使用這個TCP鏈接，減輕了服務(wù)器的壓力，提高了效率
所有的動態(tài)網(wǎng)頁開發(fā)，都必須是有狀態(tài)的協(xié)議
如果是持續(xù)連接，一個TCP是可以發(fā)送多個HTTP請求的
HTTP/1.1存在一個問題，單個TCP連接在同一時刻只能處理一個請求，意思是說： 兩個請求的生命周期不能重疊，任意兩個HTTP請求從開始到結(jié)束的時間在同一個TCP連接里不能重疊
雖然HTTP/1.1規(guī)范中規(guī)定了Pipelining來試圖解決這個問題，但此功能默認(rèn)是關(guān)閉的
Pipelining 中
客戶端可以在一個連接中發(fā)送多個請求（不需要等待任意請求的響應(yīng)）。收到請求的服務(wù)器必須按照請求收到的順序發(fā)送響應(yīng)。pipelining的缺點
1 一些代理服務(wù)器不能正確支持 HTTP pipelining
2 正確的流水線實現(xiàn)是復(fù)雜的
3 如果第一個請求的處理花費大量時間，則會導(dǎo)致后面的請求無法處理，造成阻塞。
Http2 提供了Multiplexing 多路傳輸特性，可以在一個TCP連接中同時完成多個HTTP請求
1 維持和服務(wù)其已經(jīng)建立的TCP連接，在同一個連接上順序處理多個請求
2 和服務(wù)器建立多個TCP連接瀏覽器對同一個Host 建立TCP連接數(shù)量有沒限制
Chrome 最多允許對同一個Host建立6個TCP鏈接，不同瀏覽器有區(qū)別
如果圖片都是HTTPS 連接并且在同一域名下，那么瀏覽器在SSL握手之后會和服務(wù)器協(xié)商能不能使用HTTP2，如果能的話就是用Multiplexing 功能在這個連接上進(jìn)行多路傳輸，不過也未必會所有掛載在各個域名的資源都會使用一個TCP連接獲取嗎，但可以確定的是multiplexing 可能很被用到
如果發(fā)現(xiàn)不是使用HTTP2，或者不用HTTPS，（現(xiàn)實中的 HTTP2 都是在 HTTPS 上實現(xiàn)的，所以也就是只能使用 HTTP/1.1），那么瀏覽器就會在一個HOST上建立多個TCP連接，連接數(shù)量的最大限制取決于瀏覽器的設(shè)置，這些來凝結(jié)會在空閑的時候被瀏覽器用來發(fā)送新請求，如果所有連接都在發(fā)送請求，那么其只能等待了。
同一個客戶端的兩次請求之間沒有任何關(guān)系，從服務(wù)端的角度看，他不知道這兩個請求來自同一個客戶端
最早的設(shè)計是不需要知道兩者之間的聯(lián)系的，
HTTP協(xié)議是無狀態(tài)協(xié)議
有鏈接，因為HTTP 是基于TCP 鏈接的，需要3次握手，4次斷開
詳情請看:https://blog.51cto.com/11233559/2093789
http://127.0.0.1/login?user=zhangsan&password=123
登錄窗口不能使用GET傳輸，GET頭部的長度是有限的，不能多于200多個以外的傳輸
格式是 ? 后面加key1=value1&key2=value2
當(dāng)使用POST 傳輸數(shù)據(jù)時，其相關(guān)的數(shù)據(jù)都被封裝在請求體及body中，而不是get中的直接暴露。
大的數(shù)據(jù)傳輸，必須使用POST，而不能使用GET傳輸數(shù)據(jù)。
HTML 是一種格式的約定，需要的數(shù)據(jù)是動態(tài)的，去數(shù)據(jù)庫查的數(shù)據(jù)不是死的，是動態(tài)的，靜態(tài)文本文件包括圖片
HTML 是將文本原封不動的返回，若是一個登陸的用戶名和密碼的匹配問題的時候，就不是HTML能做的事情，此時便需要動態(tài)網(wǎng)頁來完成。如python，只有腳本是不行的，這就需要類似的解釋器來進(jìn)行處理。Php,asp等動態(tài)的網(wǎng)頁技術(shù)，server page 服務(wù)器端的頁面。動態(tài)頁面中的無狀態(tài)帶來很大的問題，再次登錄將導(dǎo)致登錄后的和登錄的沒關(guān)系。既然你鏈接到我，我可以發(fā)送一個唯一標(biāo)識給你，你需要下次將這個標(biāo)識帶來，來保證是你，服務(wù)端需要發(fā)送和記錄標(biāo)識，此處需要寫入到內(nèi)存的數(shù)據(jù)結(jié)構(gòu)中，當(dāng)用戶量很大時，記錄的東西就不僅僅是這個用戶標(biāo)識了。
cookie:是一種客戶端，服務(wù)端傳遞數(shù)據(jù)的技術(shù) ,其保存的形式是鍵值對信息
瀏覽器發(fā)起每一個請求，都會把cookie信息給服務(wù)端，服務(wù)端可以通過判斷這些信息，來確定這次請求是否和之前的請求有關(guān)聯(lián)
一般來說cookie信息是在服務(wù)器端生成，返回給客戶端
客戶端可以自己設(shè)置cookie信息
Cookie 一般是當(dāng)你第一次鏈接服務(wù)器的時候服務(wù)器會查看是否有cookie帶過來，若沒有則推送一個標(biāo)識，這個標(biāo)識中會在HTTP的response包中存在，其會在瀏覽器中保存起來。如果再次對同樣網(wǎng)站發(fā)起請求，如果cookie沒過期時，其會繼續(xù)處理此標(biāo)識。若是同一個且有效，則若登錄過，則不顯示登錄頁面，若沒登錄，則強制跳轉(zhuǎn)到登錄頁面。如果一個網(wǎng)站一直登錄，其發(fā)現(xiàn)cookie快過期了，則會延長。
Cookie 是對不同的域名有區(qū)分的
cookie中加的ID 叫做session ID ，稱為會話ID，當(dāng)會話完結(jié)后，ID就消亡了，瀏覽器關(guān)閉，
Session 是存放在服務(wù)器端的，其會增加內(nèi)存。后期則使用無session， token往往中間會使用redis和memcached進(jìn)行處理
請求來的時候，其得帶著是否是同一個會話標(biāo)識
cookie可以偽造
WSGI 主要規(guī)定了服務(wù)器端和應(yīng)用程序之間的接口
瀏覽器
可以接受用戶的socket請求并和客戶端達(dá)成HTTP協(xié)議并識別解析，將數(shù)據(jù)交給后端的WSGI app 進(jìn)行處理
Server 必須支持HTTP協(xié)議，在python中實現(xiàn)了WSGI的接口，HTTP server得支持WSGI協(xié)議，將數(shù)據(jù)傳遞給程序，（app返回）然后返回給客戶端對應(yīng)的狀態(tài)情況（響應(yīng)頭），使得瀏覽器做好準(zhǔn)備，然后再返回給server，再由server將其包裝成HTTP的協(xié)議并解析處理。
后端真實處理業(yè)務(wù)的函數(shù)對象
后端APP滿足的條件
1 可通過前面的WGSI Server進(jìn)行相關(guān)的調(diào)用操作
應(yīng)用程序應(yīng)該是一個可調(diào)用對象
調(diào)用其實是回調(diào)，調(diào)用的其實是APP的某個方法
python中應(yīng)該是函數(shù)，類，實現(xiàn)了call方法的類的實例
2 這個可調(diào)用對象應(yīng)該接受兩個參數(shù)
滿足了WSGI 的基本要求，必須再留一個空，協(xié)議的封裝是需要在server端的，因此要將你寫的東西交給 http server ，由http server對返回結(jié)果進(jìn)行處理 其上述返回必須是一個可迭代對象(list,dict等)兩個參數(shù)就是入 request和出response
Handler 和 body都給了app
邏輯處理： 調(diào)用對應(yīng)的方法給客戶端。
http server 返回給app server 的參數(shù)
eviron和start_response 這兩個參數(shù)可以是任意的合法名。但一般都是這兩個名字
eviron 是包含HTTP請求信息的dict對象
| 名稱 | 含義 |
|---|---|
| REQUEST_METHOD | 請求方法，GET，PSOT，HEAD等 |
| PATH_INFO | URL 中路徑部分信息 |
| QUERY_STRING | 查詢字符串 |
| SERVER_NAME,SERVER_PORT | 服務(wù)器名，端口號 |
| HTTP_POST | 地址和端口 |
| SERVER_PROTOCOL | 協(xié)議 |
| HTTP_USER_AGENT | User Agent信息 |
start_response 是一個可調(diào)用對象，有3個參數(shù)，定義如下：
start_response(status,response_headers,exc_info=None)
status 是狀態(tài)碼。如200 ok
response_headers 是一個元素為二元祖的列表，如[('Content-Type','text/plain;charset=utf-8')]
exec_info 在錯誤處理的時候使用
start_response 應(yīng)該在返回可迭代對象之前調(diào)用，因為他返回的是Response Header，返回的可迭代對象是Response Body。
先發(fā)頭部，然后才是body
服務(wù)器端
服務(wù)器端程序需要調(diào)用符合上述定義的可調(diào)用對象，傳入environ,start_response拿到返回可迭代對象，返回給客戶端。
WSGIREF 是一個WSGI 的參考實現(xiàn)庫
wsgiref.simple_server 實現(xiàn)了一個簡單的WSGI HTTP服務(wù)器
相關(guān)參數(shù)如下
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 實現(xiàn)基本的展示頁面
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() # 關(guān)閉
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 # 響應(yīng)報文總長度
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 #暴露服務(wù)器端信息
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 # 客戶端瀏覽器可接受的類型和參數(shù)
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='<h2>Hello World</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除結(jié)果如下
環(huán)境變量數(shù)據(jù)很多，都是存儲在字典中的，字典存取沒有對象的屬性使用方便，使用第三方webob，可以把環(huán)境數(shù)據(jù)的解析，封裝成對象
pip install webob 將環(huán)境參數(shù)解析并封裝成request對象
GET方法，發(fā)送的數(shù)據(jù)是URL中的request handler中
request.get 就是一個字典MultiDict，里面就封裝著查詢字符串POST 方法，"提交"的數(shù)據(jù)是放在request body里面的，但是也同時可以使用Query String
request.POST可以獲取request Body中的數(shù)據(jù)，也是個字典MultiDict不關(guān)心什么方法提交，只關(guān)心數(shù)據(jù)，可以使用request.params，它里面是所有提交數(shù)據(jù)的封裝
#!/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) #獲取傳輸?shù)臄?shù)據(jù)，query string 或者 POST 的body
print ("method:",request.method) # 獲取請求方法
print ("path:",request.path) #獲取請求路徑
print ("user_agent:",request.user_agent) #獲取客戶端信息
print ("get data:",request.GET) #獲取get數(shù)據(jù)
print ("post data:",request.POST) # 獲取post body數(shù)據(jù)
html='<h2>Hello World</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除請求URL: http://192.168.1.200/admin/?username=mysql&password=123456
結(jié)果如下：
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)) # 此處默認(rèn)取最后一個加入的
print ('getall:',md.getall(1)) # 此處表示給據(jù)key取出所有
print (md.getone(3)) #只能有一個值，有多個值使用這個返回有問題結(jié)果如下
#!/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='<h2>Hello World</h2>'.encode("utf-8")
return [html]
server=make_server(ip,port,app) # 實例化一個websever
server.serve_forever() # 啟動
server.server_close() # 關(guān)閉
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('<h2>Hello World</h2>')
# 寫法二
#res.body='<h2>Hello Python</h2>'.encode()
#res.status_code=200
return res(environ,start_response)
server=make_server(ip,port,app) # 實例化一個websever
server.serve_forever() # 啟動
server.server_close() # 關(guān)閉
server.shutdown() # 刪除結(jié)果如下
此裝飾器傳入一個request的參數(shù)，則返回一個Response 的返回值，實現(xiàn)了一進(jìn)一出的情況
#!/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('<h2>hello python </h2>'.encode())
if __name__ == "__main__":
server = make_server(ip, port, app) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除結(jié)果如下 
什么是路由，簡單的說，就是路怎么走，就是按照不同的路徑分發(fā)數(shù)據(jù)
URL 就是不同資源的路徑，不同的路徑應(yīng)該對應(yīng)不同的應(yīng)用程序來處理，所以代碼中需要增加對路徑的處理和分析
| 路徑 | 內(nèi)容 |
|---|---|
| / | 返回歡迎內(nèi)容 |
| /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='<h2>hello World</h2>'.encode()
return res
elif request.path=="/python":
res.body='<h2>hello Python</h2>'.encode()
return res
else:
res.status_code=404
res.body='<h2>Not Found</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, app) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
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 = '<h2>hello Python</h2>'.encode()
return res
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
def show(request:Request):
res=Response()
res.status_code = 404
res.body = '<h2>Not Found</h2>'.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() # 關(guān)閉
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 = '<h2>hello Python</h2>'.encode()
return res
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
def show(request:Request):
res=Response()
res.status_code = 404
res.body = '<h2>Not Found</h2>'.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() # 關(guān)閉
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 = '<h2>hello Python</h2>'.encode()
return res
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
def show(request:Request):
res=Response()
res.status_code = 404
res.body = '<h2>Not Found</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除思想: 將需要用戶自己編寫的東西放置在類的外邊，其他的相關(guān)事件放置在類中
#!/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 = '<h2>Not Found</h2>'.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 = '<h2>hello Python</h2>'.encode()
return res
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.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() # 關(guān)閉
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 = '<h2>hello Python</h2>'.encode()
return res
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.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() # 關(guān)閉
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 = '<h2>hello World</h2>'.encode()
return res
@Application.register('/python')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除到目前為止，一個框架的雛形基本完成了
application是WSGI中的應(yīng)用程序。但是這個應(yīng)用程序已經(jīng)變成了一個路由程序，處理邏輯已移動到了應(yīng)用程序外了，而這部分就是留給程序員的部分。
目前實現(xiàn)的路由匹配，路徑匹配非常死板，使用正則表達(dá)式改造。導(dǎo)入re模塊，注冊時，存入的不再是路徑字符串，而是pattern。
__call__方法中實現(xiàn)模式和傳入路徑的比較
compile 方法，編譯正則表達(dá)式
match 方法，必須從頭開始匹配， 只匹配一次
search方法，只匹配一次
fullmath 方法，要完全匹配
findall方法，從頭開始找，找到所有匹配
字典的問題
如果使用字典，key如果是路徑，不能保證其順序，因為大多的匹配都是從嚴(yán)到寬，如果沒有一定的順序，則會導(dǎo)致問題正則表達(dá)式的預(yù)編譯問題
第一次使用會影響到用戶體驗，所以還是要在注冊的時候編譯的。綜上，改用列表，元素使用二元祖(編譯后的正則對象,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，則可以進(jìn)行下一步
return hande(request)
raise exc.HTTPNotFound('訪問資源不存在')
@Application.register('^/$')
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@Application.register('^/python$')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除請求方法，一般來說，既是是同一個URL，因為請求方法不同，處理方式也是不同的
假設(shè)一個URL。GET方法希望返回網(wǎng)頁內(nèi)容，POST方法表示瀏覽器提交數(shù)據(jù)過來需要處理并存儲進(jìn)數(shù)據(jù)庫，最終返回給客戶端存儲成功或者失敗信息，
換句話說，需要根據(jù)請求方法和正則同時匹配才能決定執(zhí)行什么樣的處理函數(shù)
| 方法 | 含義 |
|---|---|
| GET | 請求指定的頁面信息，并返回報頭和正文 |
| HEAD | 類似于get請求，只不過返回的響應(yīng)中沒有具體的內(nèi)容，用于獲取報頭 |
| POST | 向指定資源提交數(shù)據(jù)進(jìn)行處理請求(例如提交表單或者上傳文件)，數(shù)據(jù)被包含在請求正文中，POST請求可能會導(dǎo)致新的資源建立或者已有的資源的修改 |
| PUT | 從客戶端向服務(wù)器端傳遞的數(shù)據(jù)取代指定的文檔的內(nèi)容 |
| DELETE | 請求服務(wù)器刪除指定的內(nèi)容 |
#!/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 # 通過構(gòu)造方法來完成對函數(shù)的注冊
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(): # 如果請求方法和對應(yīng)注冊方法一致，則執(zhí)行對應(yīng)函數(shù)。
matcher=pattern.match(request.path)
if matcher: # 此處若能匹配到，則為True，則可以進(jìn)行下一步
return hande(request)
raise exc.HTTPNotFound('訪問資源不存在')
@Application.get('^/$')
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@Application.get('^/python$')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@Application.post('^/python$')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python POST </h2>'.encode()
return res
@Application.post('^/')
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World POST </h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除
一個URL 可以設(shè)定多種請求方法
要求：
1 如果什么方法都不寫，相當(dāng)于所有方法都支持
2 如果一個處理函數(shù)handler需要關(guān)聯(lián)多個請求方法method，如下:
@Application.register('^/$',('GET','PUT','DELETE'))
@Application.register('^/python$',('GET','PUT','DELETE'))
@Application.register('^/$',())思路：
將method變?yōu)閙ethods，將位置參數(shù)變成可變參數(shù)即可代碼如下
#!/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 # 通過構(gòu)造方法來完成對函數(shù)的注冊
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，則可以進(jìn)行下一步
return hande(request)
raise exc.HTTPNotFound('訪問資源不存在')
@Application.get('^/$')
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@Application.get('^/python')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除結(jié)果如下：
支持正則表達(dá)式的捕獲，
在框架回調(diào)__call__時，拿到request.path和正則的模式匹配后，就可以提取分組了如何處理分組?
應(yīng)用程序就是handler對應(yīng)的不同的函數(shù)，其參數(shù)request是一樣的，將捕獲的數(shù)據(jù)動態(tài)增加到request對象中即可。用動態(tài)增加屬性，為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 # 通過構(gòu)造方法來完成對函數(shù)的注冊
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，則可以進(jìn)行下一步
request.args=matcher.group() # 此處獲取元祖元素
request.kwargs=matcher.groupdict() # 此處獲取字典元素進(jìn)行處理
return hande(request)
raise exc.HTTPNotFound('訪問資源不存在')
@Application.get('^/$')
def showdefault(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@Application.get('^/python')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除所謂的路由分組，就是按照前綴分別映射
需求
URL 為 /product/123456
需要將產(chǎn)品ID提取出來
分析
這個URL可以看做是一級分組路由，生產(chǎn)環(huán)境中可以使用了
如
product=Router('/product') #匹配前綴 /product
product.get('/(?P<id>\d+)') # 匹配路徑為/product/123456
常見的一級目錄
/admin #后臺管理
/product 產(chǎn)品
這些目錄都是/跟目錄的下一級目錄，暫時稱為前綴prefix
如何建立prefix和URL 之間的隸屬關(guān)系
一個prefix下可以有若干個URL。這些URL都是屬于這個prefix中的
建立一個Router類，里面保存Prefix,同時保存URL和handler的關(guān)系以前。注冊的方法都是application的類方法，也就是所有映射信息都保存在一個類屬性中ROUTABLE中，但是現(xiàn)在要為不同的前綴就是不同的實例，因此所有注冊方法，都是實例的方法，路由包實例自己管理
application 中現(xiàn)在只需要保存所有注冊的Router對象就行了，__call__方法依然是回調(diào)入口，在其中遍歷所有的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): # 此處用于注冊二級目錄對應(yīng)的值
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，則可以進(jìn)行下一步
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) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
return response
raise exc.HTTPNotFound('訪問資源不存在')
# 注冊前綴
index=Router('/')
pyth=Router('/python')
admin=Router('/admin')
#將前綴加入對應(yīng)列表中
Application.register(pyth)
Application.register(admin)
Application.register(index)
# 寫handler
@index.get('/(\w+)')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/(\d+)')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/(\d+)')
def showadmin(request:Request):
res=Response()
res.body = '<h2>hello admin</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除通過此類，可使得kwargs這個字典，不使用[]訪問元素，使用.號訪問元素，如同屬性一樣訪問
#!/usr/bin/poython3.6
#conding:utf-8
class DictObj:
def __init__(self,d:dict): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常，此處的調(diào)用是兩個步驟，第一個是self._dict 然后會調(diào)用各種方法最終形成死循環(huán)，如果專用的字典中有的話，則其中不會訪問
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
d={
'a':1,
'b':2,
'c':3
}
x=DictOrd(d)
print (x.__dict__)
print (DictOrd.__dict__)
print (x.a)
print (x.b)結(jié)果如下
#!/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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
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): # 此處用于注冊二級目錄對應(yīng)的值
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，則可以進(jìn)行下一步
print(matcher)
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict()) # 此處通過修改后的字典，使得下面的訪問可以直接使用.來進(jìn)行訪問而不是使用[key]的方式進(jìn)行相關(guān)的訪問操作
return hande(request)
class Application:
ROUTABLE=[] # 此處修改成列表的形式比較適合順序匹配
@classmethod
def register(cls,router:Router):
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
return response
raise exc.HTTPNotFound('訪問資源不存在')
# 注冊前綴
#將前綴加入對應(yīng)列表中
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 = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/(\d+)')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/(\d+)')
def showadmin(request:Request):
res=Response()
res.body = '<h2>hello admin</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除問題
目前路由匹配使用正則表達(dá)式定義，不友好，很多用戶不會使用正則表達(dá)式，能否簡化分析
生產(chǎn)環(huán)境中。URL是規(guī)范的，不能隨意書寫，路徑是有意義的，尤其是對restful風(fēng)格，所以，要對URL規(guī)范
如 product/111102243454343 ，這就是一種規(guī)范，要求第一段是業(yè)務(wù)，第二段是ID。設(shè)計
路徑規(guī)范化，如下定義
/student/{name:str}/{id:int}
類型設(shè)計。支持str,word,int,float,any類型
通過這種定義，可以讓用戶定義簡化了，也規(guī)范了，背后的轉(zhuǎn)換是編程者實現(xiàn)的
| 類型 | 含義 | 對應(yīng)正則 |
|---|---|---|
| str | 不包含/的任意字符 | [^/]+ |
| word | 字母和數(shù)字 | \w+ |
| int | 純數(shù)字，正負(fù)數(shù) | [+-]?\d+ |
| float | 正負(fù)號，數(shù)字，包含. | [+-]?\d+.\d+ |
| any | 包含/的任意字符 | .+ |
保存類型
| 類型 | 對應(yīng)類型 |
|---|---|
| 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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(kv:str):
name,_,type=kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name,TYPEPATTERNS.get(type,'\w+')),name,TYPECAST.get(type,str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元組
def parse(src:str):
start=0
res=''
translator= {}
while True:
matcher=pattern.search(src,start) # start表示偏移量
if matcher:
res+=matcher.string[start:matcher.start()] #對匹配到的字符串進(jìn)行切割處理
tmp=transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res+=tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]]=tmp[2] # 此處保存的是名稱和類型的字典
start=matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res,translator
else: # 若不存在，也返回
return res,translator
print (parse(s))
print (parse(s1))
print (parse(s2))
print (parse(s3))
結(jié)果如下 
#!/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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
return hande(request)
class Application:
ROUTABLE=[] # 此處修改成列表的形式比較適合順序匹配
@classmethod
def register(cls,router:Router):
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
return response
raise exc.HTTPNotFound('訪問資源不存在')
# 注冊前綴
#將前綴加入對應(yīng)列表中
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 = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(request:Request):
res=Response()
res.body = '<h2>hello admin</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除處理流程
客戶端發(fā)起請求，被容器調(diào)度給Application的call
Application 中便利所有注冊的router，router通過match來判斷是否是自己處理的，通過查看request的請求來匹配注冊前綴，若符合條件，則匹配對應(yīng)的請求方法，若方法符合，則匹配對應(yīng)的URL二級目錄，并返回對應(yīng)的函數(shù)，handler處理后，返回response，applicationde拿著這個response數(shù)據(jù)，返回給原始的wsgi。
攔截器，就是要在請求處理環(huán)節(jié)的某處加入處理，有可能是中斷手續(xù)的處理
根據(jù)攔截點不同，分為:
1 請求攔截
2 響應(yīng)攔截
根據(jù)影響面分為:
1 全局?jǐn)r截
在application中攔截
2 局部攔截
在Router中攔截
相關(guān)圖形
前面的是application層面的攔截，及全局?jǐn)r截，。后面是TOUTER層面的攔截，及局部攔截，
攔截器可以是多個，多個攔截器是順序的
數(shù)據(jù)response前執(zhí)行的的命名為preinterceptor ，之后的命名為postinterceptor。
1 application 和Router 類直接加入
把攔截器的相關(guān)方法，屬性分別調(diào)價到相關(guān)的類中2 Mixin
Application 和Router類都需要這個攔截器功能，這兩個類沒什么關(guān)系，可以使用Mixin方式，將屬性，方法組合起來
但是，application類攔截器適合使用第二種方式，DNA是Router的攔截器每個實例都是不同的，所以使用第一種方式實現(xiàn)
當(dāng)出現(xiàn)多繼承時，Mixin中MRO規(guī)則會直接使用第一個，而忽略其他的__init__方法。
攔截器的函數(shù)是相對獨立的，其相當(dāng)于是相對透明的，用一個的輸出和N的輸出都應(yīng)該能夠和handler進(jìn)行處理
引入app，是為了以后從application上獲取一些全局信息，其application的實例資源。
來的輸入和輸出都是request
def fn(app,request:Request)->Request:
pass去的輸入和輸出都是response
def fn(app,request:Request,response:Response)-> Response: pass
為了把一些應(yīng)數(shù)據(jù)，配置數(shù)據(jù)，數(shù)據(jù)庫連接提供給全局共享數(shù)據(jù)提供所有對象使用，增加一個字典，存儲共享數(shù)據(jù)。將環(huán)境變量傳遞下去。
為了方便訪問，提供字典的屬性化訪問的類，因為這個字典是可寫的，和前面的類不一樣。
application最多的應(yīng)該做的是單實例模式，及就是一個實例的處理模式，若果是要用多實例，則需要使用信號量或其他進(jìn)行處理
class Context(dict): #繼承內(nèi)部類，使得類能夠提供一種能力能夠直接的屬性訪問方式，讀取配置文件的能力
def __getattr__(self, item): # 通過.的方式進(jìn)行訪問
try:
return self[item] # 自己的字典訪問方式給請求端
except KeyError: # 屬性訪問的方式導(dǎo)致的問題
raise ArithmeticError('Attribe {} Not Found'.format(item))
def __setattr__(self, key, value):
self[key]=value #處理修改和添加問題Router沒一個實例中增加上下文屬性，實例自己使用
但是Router實例如何使用全局上下文
使用新的處理方法，每一個Router實例的上下文字典內(nèi)部關(guān)聯(lián)一個全局字典的引用，如果自己的字典找不到，就去全局尋找
那Router實例什么時候關(guān)聯(lián)全局字典比較合適
在路由注冊的時候即可基本代碼如下
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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
class Context(dict): # 用于存儲共享數(shù)據(jù),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
####################上述兩種字典的不同實現(xiàn)方式處理###########################
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() # 未綁定全局的上下文，在注冊的時候進(jìn)行處理
#實例自己使用的攔截器。在match處進(jìn)行攔截
self.preinterceptor=[]
self.postinterceptor=[]
# 裝飾器需要有返回值
def reg_preinterceptor(self, fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)，裝飾器需要返回值
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
response=hande(self.ctx,request) #優(yōu)先使用自己的屬性
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前半段兩個參數(shù)，后半段三個參數(shù)
cls.PREINTERCEPTOR.append(fn)
return fn
@classmethod
def reg_postinterceptor(cls,fn):
cls.POSTINTERCEPTOR.append(fn)
return fn # 函數(shù)需要返回，其本身并沒有變動
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添加進(jìn)來，用于屬性的訪問控制及上述的NestedContext，將全局的上下文綁定給每一個router實例
# 其在router自己初始化時就自己創(chuàng)建
router.ctx.router=router #在自己的字典中中引用自己
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for fn in self.PREINTERCEPTOR: # 注冊函數(shù)，
request=fn(self.ctx,request) #第一個是全局的，第二個是自己的，定義的，需要request不變透明化
#fn(self.ctx,request) 此處此種寫法容易引起別人的誤會
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
#返回的函數(shù)進(jìn)行處理
for fn in self.POSTINTERCEPTOR: # 此處處理response相關(guān)的方法
response=fn(self.ctx.request,response)
return response
raise exc.HTTPNotFound('訪問資源不存在')
# 注冊前綴
#將前綴加入對應(yīng)列表中
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 = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(request:Request):
res=Response()
res.body = '<h2>hello admin</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除
作為一個框架，更多的功能應(yīng)該是從外部加入
1 不可能些的非常完善
2 非必要的都應(yīng)該動態(tài)加入
所以，提供一個擴展接口非常重要
#!/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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
class Context(dict): # 用于存儲共享數(shù)據(jù),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
####################上述兩種字典的不同實現(xiàn)方式處理###########################
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() # 未綁定全局的上下文，在注冊的時候進(jìn)行處理
#實例自己使用的攔截器。在match處進(jìn)行攔截
self.preinterceptor=[]
self.postinterceptor=[]
# 裝飾器需要有返回值
def reg_preinterceptor(self, fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)，裝飾器需要返回值
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
response=hande(self.ctx,request) #優(yōu)先使用自己的屬性
for fn in self.postinterceptor:
response=fn(self.ctx,request,response)
return response
class Application:
ROUTABLE=[] # 此處修改成列表的形式比較適合順序匹配
ctx=Context()
#實例的攔截器
PREINTERCEPTOR=[]
POSTINTERCEPTOR=[]
@classmethod # 增加擴展功能模塊，通過名字的方式加載進(jìn)來
def extend(cls,name,ext):
cls.ctx[name]=ext
# 攔截器的注冊
@classmethod
def reg_preinterceptor(cls,fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)
cls.PREINTERCEPTOR.append(fn)
return fn
@classmethod
def reg_postinterceptor(cls,fn):
cls.POSTINTERCEPTOR.append(fn)
return fn # 函數(shù)需要返回，其本身并沒有變動
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添加進(jìn)來，用于屬性的訪問控制及上述的NestedContext，將全局的上下文綁定給每一個router實例
# 其在router自己初始化時就自己創(chuàng)建
router.ctx.router=router #在自己的字典中中引用自己
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for fn in self.PREINTERCEPTOR: # 注冊函數(shù)，
request=fn(self.ctx,request) #第一個是全局的，第二個是自己的，定義的，需要request不變透明化
#fn(self.ctx,request) 此處此種寫法容易引起別人的誤會
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
#返回的函數(shù)進(jìn)行處理
for fn in self.POSTINTERCEPTOR: # 此處處理response相關(guān)的方法
response=fn(self.ctx.request,response)
return response
raise exc.HTTPNotFound('訪問資源不存在')
# 注冊前綴
#將前綴加入對應(yīng)列表中
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 = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(request:Request):
res=Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(request:Request):
res=Response()
res.body = '<h2>hello admin</h2>'.encode()
return res
if __name__ == "__main__":
server = make_server(ip, port, Application()) # 實例化一個websever
try:
server.serve_forever() # 啟動
except KeyboardInterrupt:
pass
finally:
server.server_close() # 關(guān)閉
server.shutdown() # 刪除在pycharm中創(chuàng)建一個包，包名為testweb
在init.py文件中，修改Application為TestWeb
通過此種方式暴露類
class TestWeb:
# 類屬性方法把類暴露出去
Router=_Router
Request=Request
Response=Response
NestedContext=NestedContext
Context=Context以供別人調(diào)用
外層新建app，將需要調(diào)用的都創(chuàng)建在app中實現(xiàn),及就是使用此模塊的人
目錄
app.py 中實現(xiàn)的代碼
from wsgiref.simple_server import make_server
from testweb import TestWeb
# 注冊前綴
#將前綴加入對應(yīng)列表中
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 = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello admin</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除testweb中_init_.py中的內(nèi)容
#!/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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
class Context(dict): # 用于存儲共享數(shù)據(jù),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
####################上述兩種字典的不同實現(xiàn)方式處理###########################
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() # 未綁定全局的上下文，在注冊的時候進(jìn)行處理
#實例自己使用的攔截器。在match處進(jìn)行攔截
self.preinterceptor=[]
self.postinterceptor=[]
# 裝飾器需要有返回值
def reg_preinterceptor(self, fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)，裝飾器需要返回值
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
response=hande(self.ctx,request) #優(yōu)先使用自己的屬性
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前半段兩個參數(shù)，后半段三個參數(shù)
cls.PREINTERCEPTOR.append(fn)
return fn
@classmethod
def reg_postinterceptor(cls,fn):
cls.POSTINTERCEPTOR.append(fn)
return fn # 函數(shù)需要返回，其本身并沒有變動
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添加進(jìn)來，用于屬性的訪問控制及上述的NestedContext，將全局的上下文綁定給每一個router實例
# 其在router自己初始化時就自己創(chuàng)建
router.ctx.router=router #在自己的字典中中引用自己
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for fn in self.PREINTERCEPTOR: # 注冊函數(shù)，
request=fn(self.ctx,request) #第一個是全局的，第二個是自己的，定義的，需要request不變透明化
#fn(self.ctx,request) 此處此種寫法容易引起別人的誤會
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
#返回的函數(shù)進(jìn)行處理
for fn in self.POSTINTERCEPTOR: # 此處處理response相關(guān)的方法
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" # 規(guī)定返回結(jié)果
response.charset='utf-8'
response.body="{}".format(content).encode() # 此處不能添加，添加了就不是json格式的數(shù)據(jù)了
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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
class Context(dict): # 用于存儲共享數(shù)據(jù),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
####################上述兩種字典的不同實現(xiàn)方式處理###########################
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() # 未綁定全局的上下文，在注冊的時候進(jìn)行處理
#實例自己使用的攔截器。在match處進(jìn)行攔截
self.preinterceptor=[]
self.postinterceptor=[]
# 裝飾器需要有返回值
def reg_preinterceptor(self, fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)，裝飾器需要返回值
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
response=hande(self.ctx,request) #優(yōu)先使用自己的屬性
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" # 規(guī)定返回結(jié)果
response.charset='utf-8'
response.body="{}".format(content).encode() # 此處不能添加，添加了就不是json格式的數(shù)據(jù)了
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前半段兩個參數(shù)，后半段三個參數(shù)
cls.PREINTERCEPTOR.append(fn)
return fn
@classmethod
def reg_postinterceptor(cls,fn):
cls.POSTINTERCEPTOR.append(fn)
return fn # 函數(shù)需要返回，其本身并沒有變動
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添加進(jìn)來，用于屬性的訪問控制及上述的NestedContext，將全局的上下文綁定給每一個router實例
# 其在router自己初始化時就自己創(chuàng)建
router.ctx.router=router #在自己的字典中中引用自己
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for fn in self.PREINTERCEPTOR: # 注冊函數(shù)，
request=fn(self.ctx,request) #第一個是全局的，第二個是自己的，定義的，需要request不變透明化
#fn(self.ctx,request) 此處此種寫法容易引起別人的誤會
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
#返回的函數(shù)進(jìn)行處理
for fn in self.POSTINTERCEPTOR: # 此處處理response相關(guān)的方法
response=fn(self.ctx.request,response)
return response
raise exc.HTTPNotFound('訪問資源不存在')app.py中的值
from wsgiref.simple_server import make_server
from testweb import TestWeb
# 注冊前綴
#將前綴加入對應(yīng)列表中
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() # 此處返回的是一個字節(jié)，需要解碼
return TestWeb.jsonify(body=body) # 此處必須傳入一個字典。否則會出問題，body是鍵，文本內(nèi)容是值
@index.get('/\w+')
def showpython(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello admin</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除1 熟悉WSGI的編程接口
2 強化模塊化，類封裝思想
3 增加分析業(yè)務(wù)的能力這個框架基本劇本了WSGI WEB 框架的基本功能，其他框架都類似。
權(quán)限驗證，SQL注入檢測的功能使用攔截器過濾。
在 testweb包外創(chuàng)建setup.py 在 testweb包內(nèi)創(chuàng)建web文件
結(jié)構(gòu)如下
web文件內(nèi)容如下
#!/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): # 將屬性中的元素添加到屬性字典中去，可能會有沖突導(dǎo)致屬性覆蓋的問題
if not isinstance(d,dict):
self.__dict__['_dict']={} # 此處不能是雙下劃綫,設(shè)置類屬性字典
else:
self.__dict__['_dict']=d #將字典加入到實例屬性列表中
def __getattr__(self, item): #此處是通過點號訪問的
try:
return self._dict[item] # 通過d.x訪問，若存在，則直接返回，若不存在，則拋出異常
except KeyError: #當(dāng)其鍵不存在的時候
raise AttributeError('Attribute {} Not Found'.format(item))
def __setattr__(self, key, value): #此處是點號修改的
# 不允許設(shè)置屬性,set表示未實現(xiàn)
raise NotImplemented
class Context(dict): # 用于存儲共享數(shù)據(jù),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
####################上述兩種字典的不同實現(xiàn)方式處理###########################
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() # 未綁定全局的上下文，在注冊的時候進(jìn)行處理
#實例自己使用的攔截器。在match處進(jìn)行攔截
self.preinterceptor=[]
self.postinterceptor=[]
# 裝飾器需要有返回值
def reg_preinterceptor(self, fn): # fn前半段兩個參數(shù)，后半段三個參數(shù)，裝飾器需要返回值
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('/({[^{}:]+:?[^{}:]*})') # 此處是提取相關(guān)用戶信息的情況,此處匹配到的只是一級目錄的相關(guān)信息
def transfrom(self,kv: str):
name, _, type = kv.strip('/{}').partition(':') # 此處用于替換操做，此處返回一個列表，通過參數(shù)解構(gòu)來收集，后面是找到第一個后進(jìn)行分割操做
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(type, '\w+')), name, self.TYPECAST.get(type,
str) # 此處的format只是構(gòu)造name和對應(yīng)正則匹配的字典，此處返回的是一個三元
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()] # 對匹配到的字符串進(jìn)行切割處理
tmp = self.transfrom(matcher.string[matcher.start():matcher.end()]) # 此處返回的是下一次匹配的結(jié)果的集合，此出返回一個三元組
res += tmp[0] # 此處保存的是名稱和正則的元組
translator[tmp[1]] = tmp[2] # 此處保存的是名稱和類型的字典
start = matcher.end() # 此處再次匹配，則需要進(jìn)行初始化繼續(xù)匹配的操做
else: # 若不能匹配，則返回
break
if res: # 若存在，則返回
return res, translator # res中保存URL，translator中保存名稱和類型的對應(yīng)關(guān)系
else: # 若不存在，也返回
return res, translator
@property
def prefix(self):
return self.__prefix
def register(self,rule,*methods): # 此處用于注冊二級目錄對應(yīng)的值
def _register(handle):
pattern,translator=self.parse(rule) #此處通過對應(yīng)的規(guī)則來處理相關(guān)配置，pattern中包含的是實際的URL路徑，translator 中包含分組名稱和對應(yīng)類型的匹配
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，則可以進(jìn)行下一步
request.args=matcher.group()
print (type(matcher.groupdict()))
request.kwargs=DictObj(matcher.groupdict())
newdict={}
for k,v in matcher.groupdict().items(): # 此處返回分組名稱和匹配值的字典，K是分組名稱，V是匹配的結(jié)果
newdict[k]=translator[k](v) #分組匹配結(jié)果,通過分組的名稱獲取對應(yīng)的類型進(jìn)行對其值進(jìn)行操作并保存
request.vars=DictObj(newdict)
response=hande(self.ctx,request) #優(yōu)先使用自己的屬性
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前半段兩個參數(shù)，后半段三個參數(shù)
cls.PREINTERCEPTOR.append(fn)
return fn
@classmethod
def reg_postinterceptor(cls,fn):
cls.POSTINTERCEPTOR.append(fn)
return fn # 函數(shù)需要返回，其本身并沒有變動
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添加進(jìn)來，用于屬性的訪問控制及上述的NestedContext，將全局的上下文綁定給每一個router實例
# 其在router自己初始化時就自己創(chuàng)建
router.ctx.router=router #在自己的字典中中引用自己
cls.ROUTABLE.append(router) # 此處用于調(diào)用上述的函數(shù)，完成數(shù)據(jù)的初始化并傳遞相關(guān)的參數(shù)prefix參數(shù)
@dec.wsgify
def __call__(self,request: Request) -> Response:
for fn in self.PREINTERCEPTOR: # 注冊函數(shù)，
request=fn(self.ctx,request) #第一個是全局的，第二個是自己的，定義的，需要request不變透明化
#fn(self.ctx,request) 此處此種寫法容易引起別人的誤會
for router in self.ROUTABLE: # 遍歷router傳輸相關(guān)參數(shù)
response=router.match(request) # 此處返回為handler的函數(shù)值
if response:
#返回的函數(shù)進(jìn)行處理
for fn in self.POSTINTERCEPTOR: # 此處處理response相關(guān)的方法
response=fn(self.ctx.request,response)
return response
raise exc.HTTPNotFound('訪問資源不存在')
if __name__ == "__main__":
pass_init_.py文件
from .web import TestWeb # 此處外部訪問只能使用TestWeb進(jìn)行各種處理，而能使用Request或Response
import json
def jsonify(**kwargs):
content=json.dumps(kwargs)
response=TestWeb.Response()
response.content_type="application/json" # 規(guī)定返回結(jié)果
response.charset='utf-8'
response.body="{}".format(content).encode() # 此處不能添加，添加了就不是json格式的數(shù)據(jù)了
return TestWeb.Response()app文件內(nèi)容
from wsgiref.simple_server import make_server
from testweb import TestWeb,jsonify
# 注冊前綴
#將前綴加入對應(yīng)列表中
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() # 此處返回的是一個字節(jié)，需要解碼
return jsonify(body=body) # 此處必須傳入一個字典。否則會出問題，body是鍵，文本內(nèi)容是值
@index.get('/\w+')
def showpython(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello World</h2>'.encode()
return res
@pyth.get('/\d+')
def showpython(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello Python</h2>'.encode()
return res
@admin.get('/\d+')
def showadmin(NestedContext,request:TestWeb.Request):
res=TestWeb.Response()
res.body = '<h2>hello admin</h2>'.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() # 關(guān)閉
server.shutdown() # 刪除setup.py 內(nèi)容
#!/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
復(fù)制到另一個環(huán)境安裝查看
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機網(wǎng)站二維碼