Python開發自定義Web框架的範例詳解

2022-07-29 14:01:06

開發自定義Web框架

接收web伺服器的動態資源請求，給web伺服器提供處理動態資源請求的服務。根據請求資源路徑的字尾名進行判斷:

如果請求資源路徑的字尾名是.html則是動態資源請求, 讓web框架程式進行處理。

否則是靜態資源請求，讓web伺服器程式進行處理。

1.開發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:
            "靜態資源請求"
            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)) + 'rn' + 
                                  'Content-Type: ' + response_type + '; charset=utf-8rn' + 
                                  'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                                  'Server: Flyme awei Serverrn'

            # 瀏覽器讀取的檔案可能不存在
            except Exception as e:
                with open('static/404.html', 'rb') as f:
                    response_body = f.read()  # 響應的主體頁面內容
                # 響應頭
                response_first_line = 'HTTP/1.1 404 Not Foundrn'
                response_header = 'Content-Length:'+str(len(response_body))+'rn' + 
                                  'Content-Type: text/html; charset=utf-8rn' + 
                                  'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                                  'Server: Flyme awei Serverrn'
            # 最後都會執行的程式碼
            finally:
                # 組成響應資料傳送給(使用者端)瀏覽器
                response = (response_first_line + response_header + 'rn').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)) + 'rn' + 
                                  'Content-Type: ' + response_type + '; charset=utf-8rn' + 
                                  'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                                  'Server: Flyme awei Serverrn'

            # 瀏覽器讀取的檔案可能不存在
            except Exception as e:
                with open('static/404.html', 'rb') as f:
                    response_body = f.read()  # 響應的主體頁面內容
                # 響應頭
                response_first_line = 'HTTP/1.1 404 Not Foundrn'
                response_header = 'Content-Length:'+str(len(response_body))+'rn' + 
                                  'Content-Type: text/html; charset=utf-8rn' + 
                                  'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                                  'Server: Flyme awei Serverrn'
            # 最後都會執行的程式碼
            finally:
                # 組成響應資料傳送給(使用者端)瀏覽器
                response = (response_first_line + response_header + 'rn').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 : 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 OKrn'
    response_header = 'Content-Length:' + str(len(response_body)) + 'rn' + 
                      'Content-Type: text/html; charset=utf-8rn' + 
                      'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                      'Server: Flyme awei Serverrn'
    response = (response_first_line + response_header + 'rn' + 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 Foundrn'
    response_header = 'Content-Length:' + str(len(response_body)) + 'rn' + 
                      'Content-Type: text/html; charset=utf-8rn' + 
                      'Date:' + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) + 'rn' + 
                      'Server: Flyme awei Serverrn'

    response = (response_first_line + response_header + 'rn').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 : 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')

小結：使用帶引數的裝飾器，可以把我們的路由自動的，新增到路由表中。

6.電影列表頁面的開發案例

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_Serverrn'

            except Exception as e:  # 瀏覽器想讀取的檔案可能不存在
                with open('static/404.html', 'rb') as f:
                    response_body = f.read()  # 響應的主體頁面內容(位元組)
                # 響應頭 （字元資料）
                response_first_line = 'HTTP/1.1 404 Not Foundrn'
                response_header = 'Server: Laoxiao_Serverrn'
            finally:
                # 組成響應資料，傳送給使用者端(瀏覽器)
                response = (response_first_line + response_header + 'rn').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 OKrn'
    response_header = 'Server: Laoxiao_Serverrn'

    response = (response_first_line + response_header + 'rn' + 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 OKrn'
    response_header = 'Server: Laoxiao_Serverrn'

    response = (response_first_line + response_header + 'rn' + 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 Foundrn'
    response_header = 'Server: Laoxiao_Serverrn'
    response = (response_first_line + response_header + 'rn').encode('utf-8') + response_body
    return response

2、根據查詢的資料得到動態的內容

到此這篇關於Python開發自定義Web框架的範例詳解的文章就介紹到這了,更多相關Python Web框架內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！