溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇“怎么使用Python開發自定義Web框架”文章的知識點大部分人都不太理解，所以小編給大家總結了以下內容，內容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“怎么使用Python開發自定義Web框架”文章吧。
接收web服務器的動態資源請求，給web服務器提供處理動態資源請求的服務。根據請求資源路徑的后綴名進行判斷:
如果請求資源路徑的后綴名是.html則是動態資源請求, 讓web框架程序進行處理。
否則是靜態資源請求，讓web服務器程序進行處理。
1、接受客戶端HTTP請求（底層是TCP）
# -*- coding: utf-8 -*-
# @File : My_Web_Server.py
# @author: Flyme awei
# @email : 1071505897@qq.com
# @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:
"靜態資源請求"
pass3、如果靜態資源怎么處理？
"靜態資源請求"
# 根據請求路徑讀取/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 : 1071505897@qq.com
# @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()1、根據請求路徑，動態的響應對應的數據
# -*- coding: utf-8 -*-
# @File : MyFramework.py
# @author: Flyme awei
# @email : 1071505897@qq.com
# @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 response2、如果請求路徑，沒有對應的響應數據也需要返回404頁面
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)
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()注意：用戶的動態資源請求，通過遍歷路由表找到對應的處理函數來完成的。
1、采用帶參數的裝飾器
# -*- coding: utf-8 -*-
# @File : My_Web_Server.py
# @author: Flyme awei
# @email : 1071505897@qq.com
# @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')小結：使用帶參數的裝飾器，可以把我們的路由自動的，添加到路由表中。
1、查詢數據
my_web.py
# -*- coding: utf-8 -*-
# @File : My_Web_Server.py
# @author: Flyme awei
# @email : 1071505897@qq.com
# @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 : 1071505897@qq.com # @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框架”這篇文章的內容，相信大家都有了一定的了解，希望小編分享的內容對大家有幫助，若想了解更多相關的知識內容，請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
