Python開發自定義Web框架的示例詳解
開發自定義Web框架
接收web服務器的動態資源請求，給web服務器提供處理動態資源請求的服務。根據請求資源路徑的後綴名進行判斷:
如果請求資源路徑的後綴名是.html則是動態資源請求, 讓web框架程序進行處理。
否則是靜態資源請求，讓web服務器程序進行處理。
1.開發Web服務器主體程序
1、接受客戶端HTTP請求（底層是TCP）
# -*- coding: utf-8 -*- # @File : My_Web_Server.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/24 21:28 from socket import * import threading # 開發自己的Web服務器主類 class MyHttpWebServer(object): def __init__(self, port): # 創建 HTTP服務的 TCP套接字 server_socket = socket(AF_INET, SOCK_STREAM) # 設置端口號互用，程序退出之後不需要等待，直接釋放端口 server_socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, True) # 綁定 ip和 port server_socket.bind(('', port)) # listen使套接字變為瞭被動連接 server_socket.listen(128) self.server_socket = server_socket # 處理請求函數 @staticmethod # 靜態方法 def handle_browser_request(new_socket): # 接受客戶端發來的數據 recv_data = new_socket.recv(4096) # 如果沒有數據，那麼請求無效，關閉套接字，直接退出 if len(recv_data) == 0: new_socket.close() return # 啟動服務器，並接受客戶端請求 def start(self): # 循環並多線程來接收客戶端請求 while True: # accept等待客戶端連接 new_socket, ip_port = self.server_socket.accept() print("客戶端ip和端口", ip_port) # 一個客戶端的請求交給一個線程來處理 sub_thread = threading.Thread(target=MyHttpWebServer.handle_browser_request, args=(new_socket, )) # 設置當前線程為守護線程 sub_thread.setDaemon(True) sub_thread.start() # 啟動子線程 # Web 服務器程序的入口 def main(): web_server = MyHttpWebServer(8080) web_server.start() if __name__ == '__main__': main()
2、判斷請求是否是靜態資源還是動態資源
# 對接收的字節數據進行轉換為字符數據
request_data = recv_data.decode('utf-8')
print("瀏覽器請求的數據:", request_data)
request_array = request_data.split(' ', maxsplit=2)
# 得到請求路徑
request_path = request_array[1]
print("請求的路徑是:", request_path)
if request_path == "/":
# 如果請求路徑為根目錄，自動設置為:/index.html
request_path = "/index.html"
# 判斷是否為：.html 結尾
if request_path.endswith(".html"):
"動態資源請求"
pass
else:
"靜態資源請求"
pass
3、如果靜態資源怎麼處理？
"靜態資源請求"
# 根據請求路徑讀取/static 目錄中的文件數據，相應給客戶端
response_body = None # 響應主體
response_header = None # 響應頭的第一行
response_first_line = None # 響應頭內容
response_type = 'test/html' # 默認響應類型
try:
# 讀取 static目錄中相對應的文件數據,rb模式是一種兼容模式，可以打開圖片，也可以打開js
with open('static'+request_path, 'rb') as f:
response_body = f.read()
if request_path.endswith('.jpg'):
response_type = 'image/webp'
response_first_line = 'HTTP/1.1 200 OK'
response_header = 'Content-Length:' + str(len(response_body)) + '\r\n' + \
'Content-Type: ' + response_type + '; charset=utf-8\r\n' + \
'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \
'Server: Flyme awei Server\r\n'
# 瀏覽器讀取的文件可能不存在
except Exception as e:
with open('static/404.html', 'rb') as f:
response_body = f.read() # 響應的主體頁面內容
# 響應頭
response_first_line = 'HTTP/1.1 404 Not Found\r\n'
response_header = 'Content-Length:'+str(len(response_body))+'\r\n' + \
'Content-Type: text/html; charset=utf-8\r\n' + \
'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \
'Server: Flyme awei Server\r\n'
# 最後都會執行的代碼
finally:
# 組成響應數據發送給(客戶端)瀏覽器
response = (response_first_line + response_header + '\r\n').encode('utf-8') + response_body
new_socket.send(response)
# 關閉套接字
new_socket.close()
靜態資源請求驗證：
4、如果動態資源又怎麼處理
if request_path.endswith(".html"):
"動態資源請求"
# 動態資源的處理交給Web框架來處理，需要把請求參數交給Web框架，可能會有多個參數，采用字典結構
params = {
'request_path': request_path
}
# Web框架處理動態資源請求後，返回一個響應
response = MyFramework.handle_request(params)
new_socket.send(response)
new_socket.close()
5、關閉Web服務器
new_socket.close()
Web服務器主體框架總代碼展示：
# -*- coding: utf-8 -*- # @File : My_Web_Server.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/24 21:28 import sys import time from socket import * import threading import MyFramework # 開發自己的Web服務器主類 class MyHttpWebServer(object): def __init__(self, port): # 創建 HTTP服務的 TCP套接字 server_socket = socket(AF_INET, SOCK_STREAM) # 設置端口號互用，程序退出之後不需要等待，直接釋放端口 server_socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, True) # 綁定 ip和 port server_socket.bind(('', port)) # listen使套接字變為瞭被動連接 server_socket.listen(128) self.server_socket = server_socket # 處理請求函數 @staticmethod # 靜態方法 def handle_browser_request(new_socket): # 接受客戶端發來的數據 recv_data = new_socket.recv(4096) # 如果沒有數據，那麼請求無效，關閉套接字，直接退出 if len(recv_data) == 0: new_socket.close() return # 對接收的字節數據進行轉換為字符數據 request_data = recv_data.decode('utf-8') print("瀏覽器請求的數據:", request_data) request_array = request_data.split(' ', maxsplit=2) # 得到請求路徑 request_path = request_array[1] print("請求的路徑是:", request_path) if request_path == "/": # 如果請求路徑為根目錄，自動設置為:/index.html request_path = "/index.html" # 判斷是否為：.html 結尾 if request_path.endswith(".html"): "動態資源請求" # 動態資源的處理交給Web框架來處理，需要把請求參數交給Web框架，可能會有多個參數，采用字典結構 params = { 'request_path': request_path } # Web框架處理動態資源請求後，返回一個響應 response = MyFramework.handle_request(params) new_socket.send(response) new_socket.close() else: "靜態資源請求" # 根據請求路徑讀取/static 目錄中的文件數據，相應給客戶端 response_body = None # 響應主體 response_header = None # 響應頭的第一行 response_first_line = None # 響應頭內容 response_type = 'test/html' # 默認響應類型 try: # 讀取 static目錄中相對應的文件數據,rb模式是一種兼容模式，可以打開圖片，也可以打開js with open('static'+request_path, 'rb') as f: response_body = f.read() if request_path.endswith('.jpg'): response_type = 'image/webp' response_first_line = 'HTTP/1.1 200 OK' response_header = 'Content-Length:' + str(len(response_body)) + '\r\n' + \ 'Content-Type: ' + response_type + '; charset=utf-8\r\n' + \ 'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \ 'Server: Flyme awei Server\r\n' # 瀏覽器讀取的文件可能不存在 except Exception as e: with open('static/404.html', 'rb') as f: response_body = f.read() # 響應的主體頁面內容 # 響應頭 response_first_line = 'HTTP/1.1 404 Not Found\r\n' response_header = 'Content-Length:'+str(len(response_body))+'\r\n' + \ 'Content-Type: text/html; charset=utf-8\r\n' + \ 'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \ 'Server: Flyme awei Server\r\n' # 最後都會執行的代碼 finally: # 組成響應數據發送給(客戶端)瀏覽器 response = (response_first_line + response_header + '\r\n').encode('utf-8') + response_body new_socket.send(response) # 關閉套接字 new_socket.close() # 啟動服務器，並接受客戶端請求 def start(self): # 循環並多線程來接收客戶端請求 while True: # accept等待客戶端連接 new_socket, ip_port = self.server_socket.accept() print("客戶端ip和端口", ip_port) # 一個客戶端的請求交給一個線程來處理 sub_thread = threading.Thread(target=MyHttpWebServer.handle_browser_request, args=(new_socket, )) # 設置當前線程為守護線程 sub_thread.setDaemon(True) sub_thread.start() # 啟動子線程 # Web 服務器程序的入口 def main(): web_server = MyHttpWebServer(8080) web_server.start() if __name__ == '__main__': main()
2.開發Web框架主體程序
1、根據請求路徑，動態的響應對應的數據
# -*- coding: utf-8 -*- # @File : MyFramework.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/25 14:05 import time # 自定義Web框架 # 處理動態資源請求的函數 def handle_request(parm): request_path = parm['request_path'] if request_path == '/index.html': # 當前請求路徑有與之對應的動態響應，當前框架隻開發瞭 index.html的功能 response = index() return response else: # 沒有動態資源的數據，返回404頁面 return page_not_found() # 當前 index函數，專門處理index.html的請求 def index(): # 需求，在頁面中動態顯示當前系統時間 data = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) response_body = data response_first_line = 'HTTP/1.1 200 OK\r\n' response_header = 'Content-Length:' + str(len(response_body)) + '\r\n' + \ 'Content-Type: text/html; charset=utf-8\r\n' + \ 'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \ 'Server: Flyme awei Server\r\n' response = (response_first_line + response_header + '\r\n' + response_body).encode('utf-8') return response def page_not_found(): with open('static/404.html', 'rb') as f: response_body = f.read() # 響應的主體頁面內容 # 響應頭 response_first_line = 'HTTP/1.1 404 Not Found\r\n' response_header = 'Content-Length:' + str(len(response_body)) + '\r\n' + \ 'Content-Type: text/html; charset=utf-8\r\n' + \ 'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + '\r\n' + \ 'Server: Flyme awei Server\r\n' response = (response_first_line + response_header + '\r\n').encode('utf-8') + response_body return response
2、如果請求路徑，沒有對應的響應數據也需要返回404頁面
3.使用模板來展示響應內容
1、自己設計一個模板 index.html ，中有一些地方采用動態的數據來替代
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>首頁 - 電影列表</title>
<link href="/css/bootstrap.min.css" rel="stylesheet">
<script src="/js/jquery-1.12.4.min.js"></script>
<script src="/js/bootstrap.min.js"></script>
</head>
<body>
<div class="navbar navbar-inverse navbar-static-top ">
<div class="container">
<div class="navbar-header">
<button class="navbar-toggle" data-toggle="collapse" data-target="#mymenu">
<span class="icon-bar"></span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
</button>
<a href="#" class="navbar-brand">電影列表</a>
</div>
<div class="collapse navbar-collapse" id="mymenu">
<ul class="nav navbar-nav">
<li class="active"><a href="">電影信息</a></li>
<li><a href="">個人中心</a></li>
</ul>
</div>
</div>
</div>
<div class="container">
<div class="container-fluid">
<table class="table table-hover">
<tr>
<th>序號</th>
<th>名稱</th>
<th>導演</th>
<th>上映時間</th>
<th>票房</th>
<th>電影時長</th>
<th>類型</th>
<th>備註</th>
<th>刪除電影</th>
</tr>
{%datas%}
</table>
</div>
</div>
</body>
</html>
2、怎麼替代，替代什麼數據
response_body = response_body.replace('{%datas%}', data)
4.開發框架的路由列表功能
1、以後開發新的動作資源的功能，隻需要：
a、增加一個條件判斷分支
b、增加一個專門處理的函數
2、路由： 就是請求的URL路徑和處理函數直接的映射。
3、路由表
| 請求路徑 | 處理函數 |
|---|---|
| /index.html | index函數 |
| /user_info.html | user_info函數 |
# 定義路由表
route_list = {
('/index.html', index),
('/user_info.html', user_info)
}
for path, func in route_list:
if request_path == path:
return func()
else:
# 沒有動態資源的數據，返回404頁面
return page_not_found()
註意：用戶的動態資源請求，通過遍歷路由表找到對應的處理函數來完成的。
5.采用裝飾器的方式添加路由
1、采用帶參數的裝飾器
# -*- coding: utf-8 -*- # @File : My_Web_Server.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/24 21:28 # 定義路由表 route_list = [] # route_list = { # ('/index.html', index), # ('/user_info.html', user_info) # } # 定義一個帶參數的裝飾器 def route(request_path): # 參數就是URL請求 def add_route(func): # 添加路由表 route_list.append((request_path, func)) @wraps(func) def invoke(*args, **kwargs): # 調用指定的處理函數，並返回結果 return func() return invoke return add_route # 處理動態資源請求的函數 def handle_request(parm): request_path = parm['request_path'] # if request_path == '/index.html': # 當前請求路徑有與之對應的動態響應，當前框架隻開發瞭 index.html的功能 # response = index() # return response # elif request_path == '/user_info.html': # 個人中心的功能 # return user_info() # else: # # 沒有動態資源的數據，返回404頁面 # return page_not_found() for path, func in route_list: if request_path == path: return func() else: # 沒有動態資源的數據，返回404頁面 return page_not_found()
2、在任何一個處理函數的基礎上增加一個添加路由的功能
@route('/user_info.html')
小結：使用帶參數的裝飾器，可以把我們的路由自動的，添加到路由表中。
6.電影列表頁面的開發案例
1、查詢數據
my_web.py
# -*- coding: utf-8 -*- # @File : My_Web_Server.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/24 21:28 import socket import sys import threading import time import MyFramework # 開發自己的Web服務器主類 class MyHttpWebServer(object): def __init__(self, port): # 創建HTTP服務器的套接字 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 設置端口號復用，程序退出之後不需要等待幾分鐘，直接釋放端口 server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) server_socket.bind(('', port)) server_socket.listen(128) self.server_socket = server_socket # 處理瀏覽器請求的函數 @staticmethod def handle_browser_request(new_socket): # 接受客戶端發送過來的數據 recv_data = new_socket.recv(4096) # 如果沒有收到數據，那麼請求無效，關閉套接字，直接退出 if len(recv_data) == 0: new_socket.close() return # 對接受的字節數據，轉換成字符 request_data = recv_data.decode('utf-8') print("瀏覽器請求的數據：", request_data) request_array = request_data.split(' ', maxsplit=2) # 得到請求路徑 request_path = request_array[1] print('請求路徑是:', request_path) if request_path == '/': # 如果請求路徑為跟目錄，自動設置為/index.html request_path = '/index.html' # 根據請求路徑來判斷是否是動態資源還是靜態資源 if request_path.endswith('.html'): '''動態資源的請求''' # 動態資源的處理交給Web框架來處理，需要把請求參數傳給Web框架，可能會有多個參數，所有采用字典機構 params = { 'request_path': request_path, } # Web框架處理動態資源請求之後，返回一個響應 response = MyFramework.handle_request(params) new_socket.send(response) new_socket.close() else: '''靜態資源的請求''' response_body = None # 響應主體 response_header = None # 響應頭 response_first_line = None # 響應頭的第一行 # 其實就是：根據請求路徑讀取/static目錄中靜態的文件數據，響應給客戶端 try: # 讀取static目錄中對應的文件數據,rb模式：是一種兼容模式，可以打開圖片，也可以打開js with open('static' + request_path, 'rb') as f: response_body = f.read() if request_path.endswith('.jpg'): response_type = 'image/webp' response_first_line = 'HTTP/1.1 200 OK' response_header = 'Server: Laoxiao_Server\r\n' except Exception as e: # 瀏覽器想讀取的文件可能不存在 with open('static/404.html', 'rb') as f: response_body = f.read() # 響應的主體頁面內容(字節) # 響應頭 （字符數據） response_first_line = 'HTTP/1.1 404 Not Found\r\n' response_header = 'Server: Laoxiao_Server\r\n' finally: # 組成響應數據，發送給客戶端(瀏覽器) response = (response_first_line + response_header + '\r\n').encode('utf-8') + response_body new_socket.send(response) new_socket.close() # 關閉套接字 # 啟動服務器，並且接受客戶端的請求 def start(self): # 循環並且多線程來接受客戶端的請求 while True: new_socket, ip_port = self.server_socket.accept() print("客戶端的ip和端口", ip_port) # 一個客戶端請求交給一個線程來處理 sub_thread = threading.Thread(target=MyHttpWebServer.handle_browser_request, args=(new_socket,)) sub_thread.setDaemon(True) # 設置當前線程為守護線程 sub_thread.start() # 子線程要啟動 # web服務器程序的入口 def main(): web_server = MyHttpWebServer(8080) web_server.start() if __name__ == '__main__': main()
MyFramework.py
# -*- coding: utf-8 -*- # @File : My_Web_Server.py # @author: Flyme awei # @email : [email protected] # @Time : 2022/7/24 21:28 import time from functools import wraps import pymysql # 定義路由表 route_list = [] # route_list = { # # ('/index.html',index), # # ('/userinfo.html',user_info) # } # 定義一個帶參數裝飾器 def route(request_path): # 參數就是URL請求 def add_route(func): # 添加路由到路由表 route_list.append((request_path, func)) @wraps(func) def invoke(*arg, **kwargs): # 調用我們指定的處理函數，並且返回結果 return func() return invoke return add_route # 處理動態資源請求的函數 def handle_request(params): request_path = params['request_path'] for path, func in route_list: if request_path == path: return func() else: # 沒有動態資源的數據，返回404頁面 return page_not_found() # if request_path =='/index.html': # 當前的請求路徑有與之對應的動態響應,當前框架，我隻開發瞭index.html的功能 # response = index() # return response # # elif request_path =='/userinfo.html': # 個人中心的功能，user_info.html # return user_info() # else: # # 沒有動態資源的數據，返回404頁面 # return page_not_found() # 當前user_info函數，專門處理userinfo.html的動態請求 @route('/userinfo.html') def user_info(): # 需求：在頁面中動態顯示當前系統時間 date = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) # response_body =data with open('template/user_info.html', 'r', encoding='utf-8') as f: response_body = f.read() response_body = response_body.replace('{%datas%}', date) response_first_line = 'HTTP/1.1 200 OK\r\n' response_header = 'Server: Laoxiao_Server\r\n' response = (response_first_line + response_header + '\r\n' + response_body).encode('utf-8') return response # 當前index函數，專門處理index.html的請求 @route('/index.html') def index(): # 需求：從數據庫中取得所有的電影數據，並且動態展示 # date = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) # response_body =data # 1、從MySQL中查詢數據 conn = pymysql.connect(host='localhost', port=3306, user='root', password='******', database='test', charset='utf8') cursor = conn.cursor() cursor.execute('select * from t_movies') result = cursor.fetchall() # print(result) datas = "" for row in result: datas += '''<tr> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s 億人民幣</td> <td>%s</td> <td>%s</td> <td>%s</td> <td> <input type='button' value='刪除'/> </td> </tr> ''' % row print(datas) # 把查詢的數據，轉換成動態內容 with open('template/index.html', 'r', encoding='utf-8') as f: response_body = f.read() response_body = response_body.replace('{%datas%}', datas) response_first_line = 'HTTP/1.1 200 OK\r\n' response_header = 'Server: Laoxiao_Server\r\n' response = (response_first_line + response_header + '\r\n' + response_body).encode('utf-8') return response # 處理沒有找到對應的動態資源 def page_not_found(): with open('static/404.html', 'rb') as f: response_body = f.read() # 響應的主體頁面內容(字節) # 響應頭 （字符數據） response_first_line = 'HTTP/1.1 404 Not Found\r\n' response_header = 'Server: Laoxiao_Server\r\n' response = (response_first_line + response_header + '\r\n').encode('utf-8') + response_body return response
2、根據查詢的數據得到動態的內容
到此這篇關於Python開發自定義Web框架的示例詳解的文章就介紹到這瞭,更多相關Python Web框架內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- python miniWeb框架搭建過程詳解
- Python 實現一個簡單的web服務器
- Python3 如何開啟自帶http服務
- 使用python+Flask實現日志在web網頁實時更新顯示
- Python Flask 請求數據獲取響應詳解