您好,登錄后才能下訂單哦!
web 也叫CS開發
CS 及客戶端,服務器端編程
客戶端,服務器端之間需要socket,約定協議,版本(往往使用的協議是TCP或UDP),指定地址和端口,就可以通信了客戶端,服務端傳輸數據,數據可以有一定的格式,雙方必須約定好
B=Browser,Browser是一種特殊的客戶端,支持HTTP(S)協議,能夠通過URL 向服務器端發起請求,等待服務端返回HTML等數據,并在瀏覽器內可視化展示程序
S=server,Server支持HTTP(S)協議,能夠接受眾多客戶端發起的HTTP請求,經過處理,將HTML等數據返回給瀏覽器
本質上來說,BS是一種特殊的CS,及客戶端必須是一種支持HTTP協議且能夠解析并渲染HTML的軟件,服務端必須是能夠接受客戶端HTTP訪問的服務器軟件
HTTP 底層使用TCP傳輸,需要通過相應的規則來處理
HTTP 超文本傳輸協議,在文本的基礎上做一些突破,相關的渲染處理,斷行等。使用標簽的方式進行規定的處理,文本已經有了一個格式,瀏覽器中必須一種能力,支持這種文本的處理和渲染。
瀏覽器必須支持HTTP協議,必須能夠理解超文本并將其繪制出來
BS 開發分為兩端開發
1 客戶端開發,或稱為前端開發
2 服務端開發,python可以學習WSGI,Django,Flask,Tornado等python WEB 框架
WSGI, web Server Gateway interface,可以看做是一種底層協議,它規定了服務器程序和應用程序各自實現什么借口,python稱為wsgirefflask: 基于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次斷開
詳情請看:https://blog.51cto.com/11233559/2093789
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='<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() # 關閉
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='<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() # 關閉
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='<h2>Hello World</h2>'.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('<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() # 關閉
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('<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() # 關閉
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='<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() # 關閉
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() # 關閉
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() # 關閉
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() # 關閉
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 = '<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() # 關閉
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() # 關閉
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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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 = '<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() # 關閉
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
復制到另一個環境安裝查看
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。