Python的web開發
1、Web開發
Tcp udpcss
Cs即客戶端、服務器端編程，客戶端和服務器端之間須要使用socket，約定協議、版本（協議使用的是tcp或者udp）。Tcp協議和udp協議，指定地址和端口，就能夠通訊了。html
客戶端、服務器端傳輸數據，數據能夠有必定的格式，雙方必須先約定好。python
一、BS
Bs：在tcp協議之上的http協議開發的，瀏覽器和服務器，是基於cs開發之上的。web
Browser、server開發正則表達式
Browser瀏覽器，一種特殊的客戶端，支持HTTP（s）協議，可以經過URL向服務器端發起請求，等待服務端返回HTML，並在瀏覽器內可視化展現的程序。編程
server，支持HTTP（s）協議，可以介紹衆多客戶端發起的HTTP協議請求，通過處理，將HTML等數據返回給瀏覽器。json
是特殊的cs，即客戶端必須是一種支持HTTP協議且能解析並渲染HTML的軟件，服務器端必須是可以接受多客戶端http訪問的軟件服務器。flask
http協議底層是基於tcp協議實現的。api
二、分類
客戶器端開發:HTML，css，Javascript瀏覽器
服務器端開發：Python有wsgi，Django，flask，tornado。
三、HTTP協議
1）簡介
安裝httpd
#yum install httpd
HTTP協議是無狀態協議。
同一個客戶端的兩次請求之間沒有任何關係，從服務器端角度來講，不知道兩個請求是來自同一個客戶端。
2）cookie
鍵值對信息，瀏覽器發起每一請求，都會把cookie信息發給服務器端，是一種客戶端、服務器端傳遞數據的技術。
服務器端能夠經過判斷這些信息，來肯定此次請求是否和以前的請求有關聯，
通常來講cookie信息是在服務器端生成的，返回給客戶端。
客戶端能夠本身設置cookie信息。
3）URL
URL就是地址，統一資源定位符uniform resource locator。每個鏈接指向一個資源供客戶端訪問。
URL組成：方言，host，path路徑，查詢字符串。
訪問靜態資源的時候，經過URL訪問的是網站某路徑下的index.html文件，而這個文件對應磁盤上的真實文件，就會從磁盤上讀取這個文件， 把這個文件發回給瀏覽器。
Scheme模式、協議：
http、ftp、https、file、mailto等等，都是相似
host：port
www.xxxx.com:80默認端口80不寫，域名會使用dns解析，
/path/to.resource
Path，指向資源的路徑。
?key1=value&key2=value2
Query string,查詢字符串，問號分割，後面的能夠=value形式，用&符號分割。
四、HTTP消息
消息分爲request和response
Request：瀏覽器向服務器端發起請求。
Response：服務器對客戶端的請求響應。
請求和響應的消息都由請求行，header消息報頭，body消息正文組成。
1）請求
（1）請求消息行：請求方法method、請求路徑、協議版本、CRLF（回車換行）。Get / http /1.1
（2）請求方法method
Get：請求獲取對應URL對應的資源。把請求的內容到放到了header的裏面。
Post：提交數據至服務器端，請求報文頭部。
Head：和get類似，不過不返回消息正文。
（3）常見傳遞信息的方式
使用get方法使用querystring
經過查詢字符串在URL中傳遞參數。
Post方法提交數據。
URL自己就包含着信息
2）響應
響應消息行：協議版本，狀態碼，消息描述CRLF（回車換行）。 http/1.1 200 ok
狀態碼在響應第一行：
1xx：提示信息，表示請求已被成功接收，繼續處理。
2xx:表示正常響應。
200.正常的返回了網頁消息
3xx:重定向
301 頁面永久移走，永久重定向，返回新的URL，瀏覽器會根據返回的URL發起新的request請求。
302 臨時重定向
304資源未修改，瀏覽器使用本地緩存。
4xx:客戶端請求錯誤
404 not found 網頁找不到，客戶端請求的資源有錯誤。
400 請求語法錯誤
401 請求要求身份驗證
403服務器拒絕請求。
5xx:服務器端錯誤
500 服務器內部錯誤
502上游服務器錯誤
五、無狀態、有鏈接和短鏈接
無狀態：服務器不知道兩次鏈接請求的之間的關係，同一個瀏覽器兩次訪問，服務器端不知道兩次之間是有聯繫的。後來能夠根據cookie，session判斷。
有鏈接：基於tcp協議，是面向鏈接的，須要3次握手，4次揮手。
短鏈接：http1.1以前，都是一個請求一個鏈接，而tcp的鏈接建立銷燬成本過高，對服務器影響很大。
從http1.1開始，支持keep-live。默認開啓，一個鏈接打開後，會保持一段時間（能夠設置），瀏覽器再次訪問該服務器就會使用這個tcp鏈接，減輕了服務器的壓力，提升了效率。
六、wsgi
Wsgi 服務器的通用網關接口。基於http協議的server。
Wsgi主要規定了服務器端和應用程序間的接口
Python Web Server Gateway Interface
通用網關接口cgi
（1）重點
基於http協議，wsgi app和server之間的應用。
對應的函數進行處理。
不一樣的路徑表明不一樣的請求。函數和請求之間的映射關係。
框架對應URL，
Server把處理邏輯提出去，提給app
瀏覽器發送請求給server，server把請求頭解包，封裝成爲字典，調用wsgi的app，app進行邏輯處理，處理完畢後直接返回狀態碼和報文頭給瀏覽器，返回正文到wsgi到server，server發送http協議正文給瀏覽器。
from wsgiref.util import setup_testing_defaults
from wsgiref.simple_server import make_server
# A relatively simple WSGI application. It's going to print out the
# environment dictionary after being updated by setup_testing_defaults
def simple_app(environ, start_response):
setup_testing_defaults(environ)
status = '200 OK'
headers = [('Content-type', 'text/plain; charset=utf-8')]
start_response(status, headers) #返回正文之間必須先返回狀態碼
ret = [("%s: %s\n" % (key, value)).encode("utf-8")
for key, value in environ.items()]
return ret
httpd = make_server('', 8000, simple_app)
print("Serving on port 8000...")
httpd.serve_forever()
七、wsgi服務器 ----wsgiref
Wsgiref.simple_server模塊實現簡單的wsgi http服務器端。
Wsgiref.Simple_server. make_server(ip,port,demo_app,server_class=wsgisercer,handler_class=wsgirequesthandler)
啓動一個wsgi服務器。
Wsgiref.simple_server.demo_app(environ,start_response)一個函數，完成了wsgi的應用程序端。
###簡單的服務器端代碼：
from wsgiref.simple_server import make_server,demo_app
ip = '127.0.0.1'
port = 9999
server = make_server(ip,port,demo_app)# demo_app應用程序，可調用
server.serve_forever()
Wsgi服務器的做用：
監聽HTTP服務端口（tcpserver，默認端口80）
接受瀏覽器端的HTTP請求並解析封裝成environ環境數據。
負責調用應用程序，將environ和start_response方法傳入。
將應用程序響應的正文封裝成http響應報文返回瀏覽器端。
八、wsgi app應用程序端
1）應用程序應該是一個可調用對象。
2）這個可調用對象接收兩個參數。
3）可調用對象，返回的是一個可迭代對象。函數和class__call__返回值是[] class中__init__ 要有yield.
def application(environ,start_response):
pass
class Application:
def __init__(self,environ,start_response):
pass
def __iter__(self):
yield self.ret
class Application:
def __call__(self, environ,start_response):
pass
app程序端三種實現：
from wsgiref.simple_server import make_server,demo_app
#
def application(environ,start_response):
status = '200 ok'
headers = [('Content-Type','text/html;charset=utf-8')]
start_response(status,headers)
ret_sre1 = b'welcome to this '
ret_sre = '{}'.format('welcome').encode()
return [ret_sre]
class Application:
def __init__(self,environ,start_response):
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status, headers)
ret_sre1 = b'welcome to this '
# ret_sre = '{}'.format('welcome').encode()
self.ret = ret_sre1
def __iter__(self):
yield self.ret
#
class Application:
def __call__(self, environ,start_response):
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status, headers)
ret_sre1 = b'welcome to this '
# ret_sre = '{}'.format('welcome').encode()
return [ret_sre1]
ip = '127.0.0.1'
port = 9999
server = make_server(ip,port,Application())# demo_app應用程序，可調用
server.serve_forever()
八、environ
是包含HTTP請求信息的dict對象
| 名稱 |
含義 |
| REQUEST_METHOD |
請求方法，get、post等 |
| PATH_INFO |
URL中的路徑部分 |
| QUERY_STRING |
查詢字符串 |
| SERVER_NAME,SERVER_POST |
服務器名、端口 |
| HTTP_HOST |
地址和端口 |
| SERVER_PROTOCOL |
協議 |
| HTTP_USER_AGENT |
Useragent信息 |
九、start_response
是一個可調用對象，有3個參數，
Status：狀態碼和描述語。
Response_headers:可迭代對象，是一個元素爲二元組的列表。[('Content-Type', 'text/html;charset=utf-8')] 響應頭部
Exc_info=None:在錯誤處理的時候使用.
start_reaponse 應該在返回可迭代對象以前調用，由於其返回的是response header，返回的可迭代對象是response body。
十、服務器端
服務器程序須要調用符合上述定義的可調用對象app，傳入environ、start_response，app處理後，返回響應頭和可迭代對象的正文，由服務器封裝返回瀏覽器端。
from wsgiref.simple_server import make_server
def application(*args):
environ = args[0]
start_response = args[1]
for k,v in environ.items():
print(k,v)
print('====================')
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status,headers)
ret = [("%s :%s\n" % (key,value)).encode('utf-8') for key,value in environ.items()]
return ret
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,application)
server.serve_forever()
測試命令：curl -I 使用head方法。
curl -X 指定方法，-d傳輸數據
十一、Web服務器
本質上就是一個tcp協議，監聽在特定的端口上
支持HTTP協議，可以將HTTP請求的報文進行解析，可以把響應數據進行HTTP協議的報文封裝並返回瀏覽器。
實現wsgi協議，協議約定了和應用程序之間接口。
HTTP協議是應用層。
十二、app應用程序
聽從wagi協議
自己是一個可調用對象
調用start_response，返回響應頭部。
返回包含正文的可迭代對象。
響應的是調用app的調用，response的是status，headers。
app的兩個參數environ和start_response
2、flask框架的實現
一、wsgi請求environment處理
Wsgi服務器會幫忙處理http請求報文，可是提供的environ是一個用起來很不方便的字典。
二、query_string查詢字符串的解析
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
print(query_string)
#1字典
# d = {}
# for item in query_string.split('&'):
# k,_,v = item.partition('=')
# d[k] = v
# print('k={},v={}'.format(k,v))
# print(d)
#2字典解析式
d = {k:v for k,_,v in map(lambda item:item.partition('='),query_string.split('&'))}
print(d)
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status,headers)
ret = 'welcome to {}'.format('item').encode()
return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
127.0.0.1 - - [26/Jun/2018 16:40:14] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
{'name': 'tome', 'age': '2'}
三、cgi模塊查詢
import cgi
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
print(query_string)
#1字典
# d = {}
# for item in query_string.split('&'):
# k,_,v = item.partition('=')
# d[k] = v
# print('k={},v={}'.format(k,v))
# print(d)
#2字典解析式
# d = {k:v for k,_,v in map(lambda item:item.partition('='),query_string.split('&'))}
# print(d)
qs = cgi.parse_qs(query_string)
print(qs)
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status,headers)
ret = 'welcome to {}'.format('item').encode()
return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
127.0.0.1 - - [26/Jun/2018 17:19:06] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
{'age': ['3', '2'], 'name': ['tome']}
Cgi模塊value的值是列表，由於value能夠有多個值。
三、urllib庫
urllib：請求解析重要的庫。
多值是age=1&age=2這個是多值。
age=1,2只是一個值。
Parse_se函數，將同一個名稱的多值，保存在字典中，使用列表保存。
import cgi
from urllib.parse import parse_qs
from wsgiref.simple_server import make_server
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
print(query_string)
#1字典
# d = {}
# for item in query_string.split('&'):
# k,_,v = item.partition('=')
# d[k] = v
# print('k={},v={}'.format(k,v))
# print(d)
#2字典解析式
# d = {k:v for k,_,v in map(lambda item:item.partition('='),query_string.split('&'))}
# print(d)
# qs = cgi.parse_qs(query_string)
# print(qs)
qs = parse_qs(query_string)
print(qs)
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status,headers)
ret = 'welcome to {}'.format('item').encode()
return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
name=tome&age=3&age=2
{'age': ['3', '2'], 'name': ['tome']}
127.0.0.1 - - [26/Jun/2018 17:28:24] "POST /xxx?name=tome&age=3&age=2 HTTP/1.1" 200 15
name=tome&age=3&age=2,3
{'age': ['3', '2,3'], 'name': ['tome']}
四、webob庫
環境數據有不少，都是存在字典中，字典的存取方式沒有對象的屬性訪問方便。
利用第三方庫webob庫，能夠把環境數據的解析、封裝成爲類。
1）webob簡介
安裝pip install webob
官方文檔：https://docs.pylonsproject.org/projects/webob/en/stable/#webob
2）webob.request對象
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print('params={}'.format(request.params))
status = '200 ok'
headers = [('Content-Type', 'text/html;charset=utf-8')]
start_response(status,headers)
ret = 'welcome to {}'.format('item').encode()
return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
1 <webob.headers.EnvironHeaders object at 0x00000050EDB88F60>
2 POST
3 /xxx
4 name=tome&age=3&age=2,3
5 GET([('name', 'tome'), ('age', '3'), ('age', '2,3')])
6 <NoVars: Not an HTML form submission (Content-Type: text/plain)>
params=NestedMultiDict([('name', 'tome'), ('age', '3'), ('age', '2,3')])
127.0.0.1 - - [26/Jun/2018 17:46:55] "POST /xxx?name=tome&age=3&age=2,3 HTTP/1.1" 200 15
3）multidict
是容許一個key存多個值 的類型
from webob.multidict import MultiDict
md = MultiDict()
md.add(1,'abc')
md.add(1,'cde')
md.add('a',1)
md.add('a',2)
for pair in md.items():
print(pair)
print(md.getall(1))
print(md.get('a'))
print(md.get(1))
(1, 'abc')
(1, 'cde')
('a', 1)
('a', 2)
['abc', 'cde']
2
Cde
4）webob.response對象
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print('params={}'.format(request.params))
res = webob.Response()
# res.status_code = 250
# print(res.content_type)
# print(res.status)
# print(res.headerlist)
# status = '200 ok'
# headers = [('Content-Type', 'text/html;charset=utf-8')]
# start_response(status,headers)
start_response(res.status,res.headerlist)
ret = 'welcome to {}'.format('item').encode('utf-8')
# res.body = ret
# return res(environ,start_response)
return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
from wsgiref.simple_server import make_server
import webob
#name=tome&age=3
def simple_app(environ,start_response):
query_string = environ.get('QUERY_STRING')
#print(query_string)
request = webob.Request(environ)
print(1,request.headers)
print(2,request.method)
print(3,request.path)
print(4,request.query_string)
print(5,request.GET)
print(6,request.POST)
print('params={}'.format(request.params))
res = webob.Response() #[('Content-Type', 'text/html;charset=utf-8')]
res.status_code = 250 #本身能夠設置屬性
print(res.content_type)
# print(res.status)
# print(res.headerlist)
# status = '200 ok'
# headers = [('Content-Type', 'text/html;charset=utf-8')]
# start_response(status,headers)
# start_response(res.status,res.headerlist)
ret = 'welcome to {}'.format('item').encode('utf-8')
res.body = ret
return res(environ,start_response) 返回res
# return [ret]
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,simple_app)
server.serve_forever()
5）webob.dec裝飾器類型
Wsgify裝飾器
https://docs.pylonsproject.org/projects/webob/en/stable/api/dec.html
from webob.dec import wsgify
import webob
from wsgiref.simple_server import make_server
@wsgify
def app(request:webob.Request):
res = webob.Response('welcome to item')
return res
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,app)
server.serve_forever()
wsgify裝飾器裝飾的函數應該具備一個參數，這個參數應該是webob。Request類型，是對字典environ的對象化後的實例。
返回值：
是一個webob. Response 類型實例。
能夠是一個bytes類型實例，會被封裝成webob. Response類型實例的body屬性。
能夠是一個字符串型實例，會被封裝轉換爲bytes類型實例，而後會被封裝成爲webob. Response類型實例的body屬性。
都會被封裝成webob.Response
一個請求，一個響應：
from webob.dec import wsgify
import webob
from wsgiref.simple_server import make_server
@wsgify
def app(request:webob.Request):
res = webob.Response('welcome to item')
return res
def application(environ,start_response):
request = webob.Request(environ)
res = webob.Response()
res.status_code = 200
ret = 'welcome'.encode('utd-8')
res.body = ret
return res(environ,start_response)
if __name__ == '__main__':
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,app)
try:
server.serve_forever()
except Exception as e:
print(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
封裝成類
from webob import Response,Request
from webob.dec import wsgify
from wsgiref.simple_server import make_server
class App:
@wsgify
def __call__(self,request:Request):
return 'welcome to'
if __name__ == '__main__':
ip = '127.0.0.1'
port = 9000
server = make_server(ip,port,App())
try:
server.serve_forever()
except Exception as e:
print(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
3、類flask框架實現
Restfull:request.post
一、路由route簡介
路由：路怎麼走，按照不一樣的路徑分發數據。
URL表明對不一樣的資源的訪問，認爲請求不一樣的路徑對應的數據。對動態網頁，不一樣的路徑對應不一樣的應用程序來處理，返回數據，用戶覺得是訪問的靜態網頁。
無論是靜態web仍是動態web服務器，都須要路徑和資源或處理程序的映射，最終返回HTML的文本。
靜態web服務器，解決路徑和文件之間的映射。
動態web服務器，解決路徑和應用程序之間的映射。
全部框架都是如此，都是由路由配置。
路由的映射關係。
from webob.dec import wsgify
from wsgiref import simple_server
from webob import Request,Response
import logging
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
def indexhandler(request):
return '<h1></h1>'
def pythonhandler(request):
return '<h2></h2>'
class App:
@wsgify
def __call__(self, request:Request):
path = request.path
if path == '/':
return indexhandler(request)
elif path == '/python':
return pythonhandler(request)
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
二、路由功能實現
路由功能，使用路由類實現。
路由類實現的功能主要就是完成path到handler函數的映射，使用字典保存合適。
理由映射創建註冊方法，兩個參數path和handler。
1)普通實現
from webob.dec import wsgify
from wsgiref import simple_server
from webob import Request,Response
import logging
from webob.exc import HTTPNotFound
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
ROUTERTABLE = {}
# @classmethod
def register(self,path,handler):
self.ROUTERTABLE[path] = handler
return handler
def indexhandler(request):
return '<h1></h1>'
def pythonhandler(request):
return '<h2></h2>'
router = Router()
router.register('/',indexhandler)
router.register('/python',pythonhandler)
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
try:
return self._Router.ROUTERTABLE.get(path)(request)
except Exception as e:
logging.info(e)
raise HTTPNotFound()
# if path == '/':
# return indexhandler(request)
# elif path == '/python':
# return pythonhandler(request)
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
2)裝飾器實現
from webob.dec import wsgify
from wsgiref import simple_server
from webob import Response,Request
import logging
from webob.exc import HTTPNotFound
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
ROUTERABLES = {}
@classmethod
def register(cls,path):
def _register(handler):
cls.ROUTERABLES[path] = handler
return handler
return _register
@Router.register('/')
def indexhandler(request):
return '<h1></h1>'
@Router.register('/python')
def pythonhandler(requset):
return '<h2></h2>'
# router = Router()
# router.register('/',indexhandler)
# router.register('/python',pythonhandler)
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
try:
return self._Router.ROUTERABLES.get(path)(request)
except Exception as e:
logging.info(e)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
三、404處理
Webob.exc提供了異常模塊
使用webob.exc.HTTPNoteFound表示路由表找不到對象的處理函數。
四、正則表達式匹配模式
使用字符串匹配的模式，路徑匹配很死板，使用正則表達式，能夠更好的匹配路徑，導入re模塊。註冊的時候傳入模式pattern。而不是字符串。
類app中實現模式和傳入的路徑 的匹配。
由於字典是無序的，在遍歷的匹配的時候須要用到有序的，因此採用列表，列表裏面的元素用二元組。（編譯後的正則對象，handler）
from webob.exc import HTTPNotFound
from webob.dec import wsgify
import logging
from webob import Request,Response
import re
from wsgiref import simple_server
class Router:
ROUTERABLE = []
@classmethod
def register(cls,pattern):
def _register(hanler):
cls.ROUTERABLE.append((re.compile(pattern),hanler))
return hanler
return _register
@Router.register('^/$')
def indexhandler(request):
return '<h1></h1>'
@Router.register('^python$')
def pythonhandler(request):
return '<h2></h2>'
class App:
_Router = Router
@wsgify
def __call__(self, request:Request):
path = request.path
for pattern,handler in self._Router.ROUTERABLE:
if pattern.match(path):
return handler(request)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
分組捕獲：
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Request,Response
import re
import logging
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern):
def _register(handler):
cls.ROUTERABLES.append((re.compile(pattern),handler))
return handler
return _register
@Rouer.register('^/$')
def indexhandler(request):
return '<h1></h1>'
@Rouer.register('^/python$')
@Rouer.register('^/python/(?P<id>\d+)$')
def pythonhandler(request):
return '<h2></h2>{}'.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for pattern,handler in self._Router.ROUTERABLES:
matcher = pattern.match(path)
if matcher:
request.groups = matcher.groups() #動態增長屬性
request.groupdict = matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
五、method方法匹配
請求方法，即便是同一個URL，由於請求方法不一樣，處理方式也不一樣。
須要按照要求匹配請求方法以後才能決定執行什麼函數。
| 方法 |
含義 |
| Get |
請求指定的頁面信息，並返回報頭和正文 |
| HEAD |
相似get請求，只不過返回的響應中沒有具體內容，用於獲取報頭 |
| Post |
向指定的資源提交數據進行出路請求（例如提交表單或者上傳文件），數據被包含在請求報文中。 Post請求客戶你能會致使新的資源創建或已有的資源修改 |
| Put |
從客戶端向服務器端傳遞的數據取代指定的文檔的內容 |
| Delete |
請求服務器刪除指定的內容 |
增長註冊屬性method方法。
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Request,Response
import re
import logging
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,method,pattern):
def _register(handler):
cls.ROUTERABLES.append((method.upper(),re.compile(pattern),handler))
return handler
return _register
@Rouer.register("GET",r'^/$')
def indexhandler(request):
return '<h1></h1>'
@Rouer.register("GET",r'^/python$')
@Rouer.register("GET",r'^/python/(?P<id>\d+)$')
def pythonhandler(request):
return '<h2></h2>{}'.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in self._Router.ROUTERABLES:
if not method == request.method.upper():
continue
matcher = pattern.match(path)
if matcher:
request.groups = matcher.groups()
request.groupdict = matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
get的方法和post方法：
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Request,Response
import re
import logging
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern,*method):
def _register(handler):
cls.ROUTERABLES.append((
tuple(map(lambda x:x.upper(),method)),
re.compile(pattern),handler))
print(cls.ROUTERABLES)
return handler
return _register
@classmethod
def get(cls,pattern):
return cls.register(pattern,"GET")
@classmethod
def post(cls,pattern):
return cls.register(pattern,"POST")
# @Rouer.register(r'^/$','GET') # 1註冊一個
# @Rouer.register(r'^/(?P<id>\d+)$') #method沒有指定，支持全部的
# @Rouer.register(r'^/$') #method沒有指定，支持全部的
# @Rouer.register(r'^/$',"GET","POST") # 1註冊多個
@Rouer.get(r'/')
def indexhandler(request):
return '<h1></h1>'
# @Rouer.register(r'^/python$','POST') #1註冊一個
# @Rouer.register(r'^/python/(?P<id>\d+)$',"GET","HEAD") #2註冊多個get head
# # @Rouer.register(r'^/python/(?P<id>\d+)$') #3不指定，all
@Rouer.post(r'/python')
def pythonhandler(request):
return '<h2></h2>{}'.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in self._Router.ROUTERABLES:
if not method or request.method.upper() in method:
matcher = pattern.match(path)
if matcher:
request.groups = matcher.groups()
request.groupdict = matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
map函數轉換的結果是惰性求值的，利用tuple。
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Request,Response
import re
import logging
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Rouer:
ROUTERABLES = []
@classmethod
def register(cls,pattern,*method):
def _register(handler):
cls.ROUTERABLES.append((
tuple(map(lambda x:x.upper(),method)),
re.compile(pattern),handler))
print(cls.ROUTERABLES)
return handler
return _register
# @Rouer.register(r'^/$','GET') # 1註冊一個
# @Rouer.register(r'^/(?P<id>\d+)$') #method沒有指定，支持全部的
# @Rouer.register(r'^/$') #method沒有指定，支持全部的
@Rouer.register(r'^/$',"GET","POST") # 1註冊多個
def indexhandler(request):
return '<h1></h1>'
# @Rouer.register(r'^/python$','POST') #1註冊一個
@Rouer.register(r'^/python/(?P<id>\d+)$',"GET","HEAD") #2註冊多個get head
# # @Rouer.register(r'^/python/(?P<id>\d+)$') #3不指定，all
def pythonhandler(request):
return '<h2></h2>{}'.format(request.groupdict)
class App:
_Router = Rouer
@wsgify
def __call__(self, request:Request):
path = request.path
for method, pattern,handler in self._Router.ROUTERABLES:
if not method or request.method.upper() in method:
matcher = pattern.match(path)
if matcher:
request.groups = matcher.groups()
request.groupdict = matcher.groupdict()
return handler(request)
raise HTTPNotFound()
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
註冊多個URL的method。
4、類flask框架
一、路由分組：
路由分組，就是按照前綴分別映射。
常見的一級目錄：
/admin後臺管理。 /product 這些目錄都是根目錄下的第一級。
兩次註冊：路由註冊，和實例的註冊。
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Request,Response
import logging
import re
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Router:
def __init__(self,predix:str=''):
self.__prefix = predix.rstrip('/\\') #前綴的
self.__routables = [] #存三元組，有序的列表
def route_register(self,pattern,*methods):
def _route_register(handler):
self.__routables.append((tuple(map(lambda x:x.upper(),methods)),
re.compile(pattern),handler))
return handler
return _route_register
def get(self,pattern):
return self.route_register(pattern,'GET')
def post(self,pattern):
return self.route_register(pattern,'POST')
def head(self,pattern):
return self.route_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,patter,handler in self.__routables:
if not method or request.method.upper() in method:
matcher = patter.match(request.path.replace(self.__prefix
,'',1))
if matcher:
request.groups = matcher.groups()
request.groupdict = matcher.groupdict()
return handler(request)
class App:
_ROUTES = []
@classmethod
def register(cls,*routers:Router):
for router in routers:
cls._ROUTES.append(router)
@wsgify
def __call__(self,request):
for router in self._ROUTES:
response = router.match(request)
if response:#匹配返回非None的router對象
return response #匹配則返回
raise HTTPNotFound()
idx = Router()
py = Router('/python')
App.register(idx,py)
@idx.get(r'^/$')
@idx.route_register(r'^/(?P<id>\d+)$')
def indexhandler(request):
return '<h1></h1>'
@py.get('^/(\w+)$')
@py.route_register(r'^/(?P<id>\d+)$')
def pythonhandler(request):
return '<h2></h2>'
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
二、字典的屬性方法
屬性變成字典的方式來進行訪問。
讓kwargs這個字典，不使用[]訪問元素，而是使用.點號訪問元素。如同屬性同樣訪問。
class AttrDict:
def __init__(self,d):
self.__dict__.update(d)
def __setattr__(self, key, value):
raise NotImplementedError
d = {'a':1,'b':2}
obj = AttrDict(d)
print(obj.a)
加上屬性字典：
from webob.exc import HTTPNotFound
from webob.dec import wsgify
from webob import Request,Response
import re
from wsgiref import simple_server
import logging
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class AttrDict:
def __init__(self,d:dict):
self.__dict__.update(d if isinstance(d,dict)else {})
def __setattr__(self, key, value):
raise NotImplementedError
def __repr__(self):
return '<AttrDict{}>'.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class Router:
def __init__(self,prefix:str=''):
self._prefix = prefix.rstrip('\\/') #前綴
self.routerables = [] #三元組
def router_ables(self,pattern,*methods):
def _router_ables(handler): #三元組
self.routerables.append((
tuple(map(lambda x:x.upper(),methods)),
re.compile(pattern),handler))
return handler
return _router_ables
def get(self,pattern):
return self.router_ables(pattern,"GET")
def post(self,pattern):
return self.router_ables(pattern,"POST")
def head(self,pattern):
return self.router_ables(pattern,"HEAD")
def matcher(self,request:Request):
if not request.path.startswith(self._prefix):
return None
for method,pattern,handler in self.routerables:
if not method or request.method.upper() in method:
matcher = pattern.match(request.path.replace(self._prefix,'',1))
if matcher: #匹配上
request.groups = matcher.groups()
request.groupdict = AttrDict(matcher.groupdict()) #命名分組的字典被屬性化
print(request.groupdict.id)
return handler(request)
class App: #一級註冊函數
_ROUTERS = []
@classmethod
def app_register(cls,*routers:Router):
for router in routers:
cls._ROUTERS.append(router)
@wsgify
def __call__(self, request:Request):
for router in self._ROUTERS:
response = router.matcher(request) #前綴匹配
if response:
return response #匹配上返回，沒返回直接報錯404
raise HTTPNotFound()
idx = Router()
py = Router('/python')
App.app_register(idx,py)
@idx.get(r'^/$')
@idx.router_ables(r'^/(?P<id>\d+)$')
def indexhanler(request):
return '<h1></h1>'
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@idx.get(r'^/(\w+)$')
@py.router_ables(r'^/(?P<id>\d+)$')
def pythonhandler(request):
return '<h2></h2>{}'.format(request.groupdict.id)
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except Exception as e:
logging.info(e)
except KeyboardInterrupt:
server.shutdown()
server.server_close()
三、正則表達式化簡
| 類型 |
含義 |
對應的正則 |
| str |
不包含/的任意字符，默認類型 |
[^/]+ |
| word |
字母和數字 |
\w+ |
| int |
純數字，正負數 |
[+-]?\d+ |
| float |
正負號，數字，包含 |
[+-]?\d+.\d+ |
| any |
包含/的任意字符。 |
.+ |
類型映射到正則表達式；就是使用format函數的填空模式。
TYPEPATTERNS = {
'str' : r'[^/]+',
'word' : r'\w+',
'int' : r'[+-]?\d+',
'float': r'[+-]?\d+.\d+',
'any' : r'.+'
}
name = 'id'
t = 'int'
print('/(?P<{}>{})'.format(name,TYPEPATTERNS.get(t,TYPEPATTERNS['str'])))
捕獲匹配用戶的輸入，捕獲到分組名，而後對應正則表達式。
src = '/student/{name:str}/{id:int}'
import re
# pattern = r'/{(\w+):(\w+)}'
pattern = r'/{([^{}:]*):([^{}:]*)}'
regix = re.compile(pattern)
matcher = regix.search(src)
def repl(mathcer):
name = matcher.group(1)
t = matcher.group(2)
return '/(?P<{}>{})'.format(name,TYPEPATTERNS.get(t,TYPEPATTERNS['str']))
# print(matcher.group(1))
# print(matcher.group(2))
print(regix.sub(repl,src))
根據用戶輸入/{id:int}/{name:str} 把這些信息轉化爲對應的正則表達式。
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Response,Request
import logging
import re
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class AttrDict:
def __init__(self,d:dict):
self.__dict__.update(d if isinstance(d,dict)else {})
def __setattr__(self, key, value):
raise NotImplementedError
def __repr__(self):
return '<AttrDict{}>'.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r'/{([^{}:]+):?([^{}:]*)}'
_regex = re.compile(pa)
TYPEPATTERNS = {
'str': r'[^/]+',
'word': r'\w+',
'int': r'[+-]?\d+',
'float': r'[+-]?\d+.\d+',
'any': r'.+'
}
def repl(self,matcher):
name = matcher.group(1)
t = matcher.group(2)
return '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
def pase(self,src):
return self._regex.sub(self.repl,src)
def __init__(self,prefix:str=''):
self.__prefix = prefix.rstrip('\\/')
self.__ROUTERABLES = []
def router_register(self,rule,*method):
def _router_register(handler):
self.__ROUTERABLES.append(
(tuple(map(lambda x:x.upper(),method)),
re.compile(self.pase(rule)),handler))
print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,'GET')
def post(self,pattern):
return self.router_register(pattern,'POST')
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,handler in self.__ROUTERABLES:
if not method or request.method.upper() in method:
matcher = pattern.match(request.path.replace(self.__prefix,'',1))
if matcher:
# request.groups = matcher.groups()
# print(2,request.groups)
request.groupdict = AttrDict(matcher.groupdict())
print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable in self._ROUTER:
response = routerable.match(request)
if response:
return response
raise HTTPNotFound()
idx = RouterAbles('/')
py = RouterAbles('/python')
App.app_regsieter(idx,py)
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
@idx.router_register(r'^/{id:int}$')
def handler(request):
id = ''
if request.groupdict:
id = request.groupdict.id
return '<h1></h1>{}'.format(request.groupdict.id)
# @py.get(r'^/$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@py.router_register(r'^/{id}$')
def pythonhandler(request):
if request.groupdict:
id = request.groupdict.id
return '<h2></h2>{}'.format(id)
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
訪問方式：
127.0.0.1:9000/python/1234
127.0.0.1:9000/123
上例中缺點是捕獲的信息sub函數捕獲的所有爲str。轉化的話。須要改造sub函數。
方法：一、路由註冊的時候，將用戶指定的rule（@py.router_register(r'^/{id}$')）添加入路由表三元組中。添加的時候調用parse函數。
二、parse函數中的返回值爲：模式匹配sub匹配用戶rule中的name和類型，sub（兩個參數，調用另一個函數depl，rule）。
三、depl函數中，根據信息。提早指定的表格，按照名字和類型，將類型轉化爲對應的正則表達式類型。
用戶輸入三種的形式的匹配，沒寫的話採用默認值。
拼接字符串的形式：
改造parse函數的代碼
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Response,Request
import logging
import re
from wsgiref import simple_server
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class AttrDict:
def __init__(self,d:dict):
self.__dict__.update(d if isinstance(d,dict)else {})
def __setattr__(self, key, value):
raise NotImplementedError
def __repr__(self):
return '<AttrDict{}>'.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r'/{([^{}:]+):?([^{}:]*)}'
_regex = re.compile(pa)
TYPEPATTERNS = {
'str': r'[^/]+',
'word': r'\w+',
'int': r'[+-]?\d+',
'float': r'[+-]?\d+.\d+',
'any': r'.+'
}
# def repl(self,matcher):
# print(matcher,']]]')
# name = matcher.group(1)
# t = matcher.group(2)
# s1 = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
# print('----',s1)
# return s1
#
# def pase(self,rule):
# s2= self._regex.sub(self.repl,rule)
# print('+-+-',rule)
# print('=====',s2)
# return s2
def parse(self,src: str):
start = 0
repl = ''
types = {}
TYPECAST = {
'str': str,
'word': str,
'int': int,
'float': float,
'any': str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl += src[start:matcher.start()]
tmp = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=''):
self.__prefix = prefix.rstrip('\\/')
self.__ROUTERABLES = []
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types = self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda x:x.upper(),method)),
re.compile(pattern),types,handler))
print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,'GET')
def post(self,pattern):
return self.router_register(pattern,'POST')
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method or request.method.upper() in method:
matchers = pattern.match(request.path.replace(self.__prefix,'',1))
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] = trans[k](v)
# request.groups = matcher.groups()
# print(2,request.groups)
request.vars = AttrDict(matchers.groupdict())
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable in self._ROUTER:
response = routerable.match(request)
if response:
return response
raise HTTPNotFound()
idx = RouterAbles('/')
py = RouterAbles('/python')
App.app_regsieter(idx,py)
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
@idx.router_register(r'^/{id:int}$')
def handler(request):
id = ''
if request.vars:
id = request.vars.id
return '<h1></h1>{}'.format(id)
# @py.get(r'^/$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@py.router_register(r'^/{id}$')
def pythonhandler(request):
if request.vars:
id = request.vars.id
return '<h2></h2>{}'.format(id)
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
app中，可使用字典保存全部的router實例。由於每個router的實例的前綴不一樣，徹底使用前綴爲key，router實例爲value組成字典。，之後在call方法中，就不須要遍歷列表了。只是須要提取request的path的前綴就能夠和字典的key匹配了。提升了效率。
四、框架處理流程
客戶端發來http請求，被wsgi服務器處理後傳遞給app的__call__方法。
App中遍歷已經註冊的router ，router的match來判斷本身能不能處理，前綴匹配，看到註冊的規則。（規則是裝飾器已經轉換成了命名分組的正則表達式了）
若是由某一個註冊的正則表達式匹配，就把回去的參數放到request中，調用註冊時候的映射handler傳入request。
Handler處理之後，返回response。App中拿到response的數據，返回給wsgi。
Handler返回的僅僅是數據，將數據填入HTML中，將新生成的HTML字符串返回給客戶端，就是網頁技術。模板技術。
5、flask實現
一、模板
HTML：就是格式和數據混在一塊兒的超文本。數據格式的描述。
XML：數據描述。
動態數據很難緩存。緩存是沒有用的。
import re
from io import StringIO,BytesIO
d = {'id':5,'name':'tom','age':20}
class Template:
_pattern = '{{([a-zA-Z0-9_]+)}}'
regex = re.compile(_pattern)
@classmethod
def render(cls,template,data:dict):
html = StringIO()
with open(template,encoding='utf-8')as f:
for line in f :
start = 0
newline = ''
for matcher in cls.regex.finditer(line):
newline += line[start:matcher.start()]
print(matcher,matcher.group(1))
key = matcher.group(1)
tmp = data.get(key,'')
newline += str(tmp)
start = matcher.end()
else:
newline += line[start:]
html.write(newline)
print(html.getvalue())
html.close()
filename = 'index.html'
Template.render(filename,d)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>wcl</title>
</head>
<body>
顯示數據<br>
{{id}}{{name}}{{age}}
</body>
</html>
渲染後的代碼：
2 <_sre.SRE_Match object; span=(3, 9), match='{{id}}'> id
2 <_sre.SRE_Match object; span=(9, 17), match='{{name}}'> name
2 <_sre.SRE_Match object; span=(17, 24), match='{{age}}'> age
1 <!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>wcl</title>
</head>
<body>
<h1>顯示數據</h1>
<p>5tom20</p>
</body>
</html>
二、jinja
1）簡介
基於模塊的引擎。設計思想來自Django的模板引擎，與其及其類似。
文檔：http://jinja.pocoo.org/docs/2.10/
http://docs.jinkan.org/docs/jinja2/
2）安裝
安裝：pip install jinjia2
pip install Markupsafe
3）模板構建
當前目錄下，新建包webarch，下面闖進一個目錄templates，該目錄下新建一個HTML模板文件index.html
經常使用的模式
模板代碼：
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
顯示數據<br>
<ul>
{% for id,name,age in userlist %}
<li>{{loop.index}}{{id}} ,{{name}},{{age}}</li>li>
{% endfor %}
</ul>
total{{usercount}}ren
</body>
</html>
Template文件代碼：
from jinja2 import Environment ,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader('webarch','templates'),
autoescape=select_autoescape(['html','xml'])
)
# env1 = Environment(loader=FileSystemLoader('templates'))
# d = {'id':5,
# 'name':'tom',
# 'age':20,
# 'list':[]
# }
# d = {
# 'userlist':[
# (1,'tom',20),
# (3,'jeery',23),
# (7,'ll',28)
# ]
# }
def render(name,data:dict):
template = env.get_template('index.html')
html = template.render(**data)
return html
三、json
from .web import RouterAbles,Request,Response,App
from .template import render
idx = RouterAbles('/')
py = RouterAbles('/python')
App.app_regsieter(idx,py)
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
@idx.router_register(r'^/{id:int}$')
def handler(request):
userlist = [
(3,'tom',20),
(5,'jerry',23),
(6,'ll',25),
(7,'ls',30)
]
d = {'userlist':userlist,'usercount':len(userlist)}
# res = Response(json=d) #json支持
# return res
return render('index.html',d)
# id = ''
# if request.vars:
# id = request.vars.id
# return '<h1></h1>{}'.format(id)
# @py.get(r'^/$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@py.router_register(r'^/{id}$')
def pythonhandler(request):
userlist = [
(3, 'tom', 20),
(5, 'jerry', 23),
(6, 'll', 25),
(7, 'ls', 30)
]
d = {'userlist': userlist, 'usercount': len(userlist)}
return render('index.html', d)
# res = Response(json=d) #json支持
# return res
# if request.vars:
# id = request.vars.id
# return '<h2></h2>{}'.format(id)
四、模塊化
將全部的代碼組織成包和模塊。
1.app.py
2.webarch(package)
(1)__init__.py
(2)template.py
(3)web.py
(4)templates(package)
Index.html
完整代碼：
（1）app.py （server）
from wsgiref import simple_server
from webarch import App
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
（2）__init__.py (handler函數)
from .web import RouterAbles,Request,Response,App
from .template import render
idx = RouterAbles('/')
py = RouterAbles('/python')
App.app_regsieter(idx,py)
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
@idx.router_register(r'^/{id:int}$')
def handler(request):
userlist = [
(3,'tom',20),
(5,'jerry',23),
(6,'ll',25),
(7,'ls',30)
]
d = {'userlist':userlist,'usercount':len(userlist)}
return render('index.html',d)
# id = ''
# if request.vars:
# id = request.vars.id
# return '<h1></h1>{}'.format(id)
# @py.get(r'^/$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@py.router_register(r'^/{id}$')
def pythonhandler(request):
userlist = [
(3, 'tom', 20),
(5, 'jerry', 23),
(6, 'll', 25),
(7, 'ls', 30)
]
d = {'userlist': userlist, 'usercount': len(userlist)}
return render('index.html', d)
# if request.vars:
# id = request.vars.id
# return '<h2></h2>{}'.format(id)
（3）template.py （渲染模塊加載文件）
from jinja2 import Environment ,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader('webarch','templates'),
autoescape=select_autoescape(['html','xml'])
)
# env1 = Environment(loader=FileSystemLoader('templates'))
# d = {'id':5,
# 'name':'tom',
# 'age':20,
# 'list':[]
# }
# d = {
# 'userlist':[
# (1,'tom',20),
# (3,'jeery',23),
# (7,'ll',28)
# ]
# }
def render(name,data:dict):
"""
:param name: 去模塊目錄搜索此模板名的文件
:param data: 字典
:return: 字符串
"""
template = env.get_template('index.html')
html = template.render(**data)
return html
（4）web.py （app，routerable，attrdict類）
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Response,Request
import logging
import re
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class AttrDict:
def __init__(self,d:dict):
self.__dict__.update(d if isinstance(d,dict)else {})
def __setattr__(self, key, value):
raise NotImplementedError
def __repr__(self):
return '<AttrDict{}>'.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r'/{([^{}:]+):?([^{}:]*)}'
_regex = re.compile(pa)
TYPEPATTERNS = {
'str': r'[^/]+',
'word': r'\w+',
'int': r'[+-]?\d+',
'float': r'[+-]?\d+.\d+',
'any': r'.+'
}
# def repl(self,matcher):
# print(matcher,']]]')
# name = matcher.group(1)
# t = matcher.group(2)
# s1 = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
# print('----',s1)
# return s1
#
# def pase(self,rule):
# s2= self._regex.sub(self.repl,rule)
# print('+-+-',rule)
# print('=====',s2)
# return s2
def parse(self,src: str):
start = 0
repl = ''
types = {}
TYPECAST = {
'str': str,
'word': str,
'int': int,
'float': float,
'any': str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl += src[start:matcher.start()]
tmp = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=''):
self.__prefix = prefix.rstrip('\\/')
self.__ROUTERABLES = []
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types = self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda x:x.upper(),method)),
re.compile(pattern),types,handler))
#print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,'GET')
def post(self,pattern):
return self.router_register(pattern,'POST')
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method or request.method.upper() in method:
matchers = pattern.match(request.path.replace(self.__prefix,'',1))
#print('+++',trans)
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] = trans[k](v)
#print('+-+-',trans[k](v))
#print(1,matchers.groupdict().items())
#print(k,v)
#print('---',newdict)
# request.groups = matcher.groups()
# print(2,request.groups)
request.vars = AttrDict(newdict)
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
for routerable in self._ROUTER:
response = routerable.match(request)
if response:
return response
raise HTTPNotFound()
（5）index.html文件內容，渲染模板
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
顯示數據<br>
<ul>
{% for id,name,age in userlist %}
<li>{{loop.index}}{{id}} ,{{name}},{{age}}</li>li>
{% endfor %}
</ul>
total{{usercount}}ren
</body>
</html>
五、攔截器interceptor
攔截器是在請求處理的環節的某一處加入處理，有多是中斷後續的處理
分類：鏈式、洋蔥式的。
攔截點不一樣：
1）請求時候攔截的
2）響應時候攔截
影響面：
1）全局攔截，在app中攔截
2）局部攔截，在router中攔截
攔截器能夠是多個，多個攔截器是有順序的，數據的response前執行的命名爲preinterceptor，以後命名爲postinterceptor。
加入攔截器功能的方式
（1）app和router類直接加入。
把攔截器相關方法，屬性分別添加到相關的類中。
實現簡單
（2）Mixin
App和router類都須要這個攔截器功能，這個兩個類沒有什麼關係，可使用Mixin方式，將屬性、方法組合進來。App攔截器適合使用第二種，可是router的攔截器是每一個實例不同的，因此使用第一種方式實現。
攔截器函數的設計，fn不能影響數據向下一級的傳遞，也就是透明的。
Ip攔截的簡單代碼;
from webob.exc import Request
def ip(request:Request):
if request.remote_addr.startswish('192'):
return request
else:
return None
六、發佈的模式
from distutils.core import setup
import glob
setup(name='webarch', # 名字
version='0.0.1', #版本
description='python wsgi web framework', #描述信息
author='wcl', #做者
author_email='604603701@qq.com', #做者郵箱
url='www.', #包的主頁
packages=['webarch'], #打包列表，
data_files=glob.glob('webarch/templates/*') #返回列表 配置文件，圖片等文件列表
)
#打包
##python setup.py sdist
# 安裝
#pip install webarch
七、總結
（1）熟悉wsgi的編程接口
（2）強化模塊、類封裝的思想
（3）加強業務分析的能力。
框架基本具有了wsgi web框架 的基本實現。
權限驗證，SQL注入的檢測功能都須要使用攔截器功能實現
完整代碼：
（1）__init__.py文件
from .web import RouterAbles,Request,Response,App
from .template import render
idx = RouterAbles('/')
py = RouterAbles('/python')
App.app_regsieter(idx,py)
def ip(request:Request):
if request.remote_addr.startswish('192'):
return request
else:
return None
py.register_preinterceprot(ip)
# @idx.get(r'^/$')
# @idx.router_register(r'^/(?P<id>\d+)$')
@idx.router_register(r'^/{id:int}$')
def handler(request):
userlist = [
(3,'tom',20),
(5,'jerry',23),
(6,'ll',25),
(7,'ls',30)
]
d = {'userlist':userlist,'usercount':len(userlist)}
# res = Response(json=d) #json支持
# return res
return render('index.html',d)
# id = ''
# if request.vars:
# id = request.vars.id
# return '<h1></h1>{}'.format(id)
# @py.get(r'^/$')
# @py.get('^/(\w+)$')
# @py.router_register(r'^/(?P<id>\d+)$')
@py.router_register(r'^/{id}$')
def pythonhandler(request):
userlist = [
(3, 'tom', 20),
(5, 'jerry', 23),
(6, 'll', 25),
(7, 'ls', 30)
]
d = {'userlist': userlist, 'usercount': len(userlist)}
return render('index.html', d)
# res = Response(json=d) #json支持
# return res
# if request.vars:
# id = request.vars.id
# return '<h2></h2>{}'.format(id)
（2）類文件web.py
from webob.dec import wsgify
from webob.exc import HTTPNotFound
from webob import Response,Request
import logging
import re
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class AttrDict:
def __init__(self,d:dict):
self.__dict__.update(d if isinstance(d,dict)else {})
def __setattr__(self, key, value):
raise NotImplementedError
def __repr__(self):
return '<AttrDict{}>'.format(self.__dict__)
def __len__(self):
return len(self.__dict__)
class RouterAbles:
pa = r'/{([^{}:]+):?([^{}:]*)}'
_regex = re.compile(pa)
TYPEPATTERNS = {
'str': r'[^/]+',
'word': r'\w+',
'int': r'[+-]?\d+',
'float': r'[+-]?\d+.\d+',
'any': r'.+'
}
# def repl(self,matcher):
# print(matcher,']]]')
# name = matcher.group(1)
# t = matcher.group(2)
# s1 = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
# print('----',s1)
# return s1
#
# def pase(self,rule):
# s2= self._regex.sub(self.repl,rule)
# print('+-+-',rule)
# print('=====',s2)
# return s2
def parse(self,src: str):
start = 0
repl = ''
types = {}
TYPECAST = {
'str': str,
'word': str,
'int': int,
'float': float,
'any': str
}
matchers = self._regex.finditer(src)
for i, matcher in enumerate(matchers):
name = matcher.group(1)
t = matcher.group(2)
types[name] = TYPECAST.get(t, str)
repl += src[start:matcher.start()]
tmp = '/(?P<{}>{})'.format(name, self.TYPEPATTERNS.get(t, self.TYPEPATTERNS['str']))
repl += tmp
start = matcher.end()
else:
repl += src[start:]
return repl, types
def __init__(self,prefix:str=''):
self.__prefix = prefix.rstrip('\\/')
self.__ROUTERABLES = []
#攔截器
self.pre_interceptor = []
self.post_interceptor = []
def register_preinterceprot(self,fn):
self.pre_interceptor.append(fn)
return fn
def register_postinterport(self,fn):
self.post_interceptor.append(fn)
return fn
def router_register(self,rule,*method):
def _router_register(handler):
pattern ,types = self.parse(rule)
self.__ROUTERABLES.append(
(tuple(map(lambda x:x.upper(),method)),
re.compile(pattern),types,handler))
#print(1,self.__ROUTERABLES)
return handler
return _router_register
def get(self,pattern):
return self.router_register(pattern,'GET')
def post(self,pattern):
return self.router_register(pattern,'POST')
def head(self,pattern):
return self.router_register(pattern,"HEAD")
def match(self,request:Request):
if not request.path.startswith(self.__prefix):
return None
#請求攔截，處理request
for fn in self.pre_interceptor:
request = fn(request)
if not request:
return None
for method,pattern,trans,handler in self.__ROUTERABLES:
if not method or request.method.upper() in method:
matchers = pattern.match(request.path.replace(self.__prefix,'',1))
#print('+++',trans)
if matchers:
newdict ={}
for k,v in matchers.groupdict().items():
newdict[k] = trans[k](v)
#print('+-+-',trans[k](v))
#print(1,matchers.groupdict().items())
#print(k,v)
#print('---',newdict)
# request.groups = matcher.groups()
# print(2,request.groups)
request.vars = AttrDict(newdict)
# print(3,request.groupdict)
return handler(request)
class App:
_ROUTER = []
#全局攔截
PRE_INTERCEPTOR = []
POST_INTERCEPTOR = []
#全局攔截器註冊函數
@classmethod
def register_preinterceptor(cls,fn):
cls.PRE_INTERCEPTOR.append(fn)
return fn
@classmethod
def register_postinterceptor(cls,fn):
cls.POST_INTERCEPTOR.append(fn)
@classmethod
def app_regsieter(cls,*routerables:RouterAbles):
for routerable in routerables:
cls._ROUTER.append(routerable)
print(4,cls._ROUTER)
@wsgify
def __call__(self, request):
#全局請求攔截
for fn in self.PRE_INTERCEPTOR:
request = fn(request)
#全局攔截響應
for routerable in self._ROUTER:
response = routerable.match(request)
for fn in self.POST_INTERCEPTOR:
response = fn(request,response)
if response:
return response
raise HTTPNotFound()
（3）template.py文件
from jinja2 import Environment ,PackageLoader,select_autoescape,FileSystemLoader
env = Environment(
loader=PackageLoader('webarch','templates'),
autoescape=select_autoescape(['html','xml'])
)
# env1 = Environment(loader=FileSystemLoader('templates'))
# d = {'id':5,
# 'name':'tom',
# 'age':20,
# 'list':[]
# }
# d = {
# 'userlist':[
# (1,'tom',20),
# (3,'jeery',23),
# (7,'ll',28)
# ]
# }
def render(name,data:dict):
"""
:param name: 去模塊目錄搜索此模板名的文件
:param data: 字典
:return: 字符串
"""
template = env.get_template('index.html')
html = template.render(**data)
return html
（4）app.py 文件
from wsgiref import simple_server
from webarch import App
if __name__ == '__main__':
server = simple_server.make_server('127.0.0.1',9000,App())
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
（5）index.html
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> 顯示數據<br> <ul> {% for id,name,age in userlist %} <li>{{loop.index}}{{id}} ,{{name}},{{age}}</li>li> {% endfor %} </ul> total{{usercount}}ren </body> </html>
- 1. Python Web開發
- 2. python-web開發
- 3. python之Web開發
- 4. [轉載]Python Web 開發:使用 Django 和 Python 開發WEB 站點
- 5. 淺談Java Web開發與Python Web開發的區別
- 6. Flask Web開發：基於Python的Web應用開發實戰
- 7. Python Web開發：開發wsgi中間件
- 8. Python中WEB開發（一）
- 9. Python Web開發之Flask
- 10. Head First Python 7 web開發
- 更多相關文章...
- • SQLite - Python - SQLite教程
- • Docker 安裝 Python - Docker教程
- • PHP開發工具
- • JDK13 GA發佈：5大特性解讀
-
每一个你不满意的现在，都有一个你没有努力的曾经。