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 : 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)) + '\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()

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 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を自分で設計し、動的データを使用してそれを置き換えます。 places

<!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。 htmlindex function/user_info.htmluser_info function

# 定义路由表
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_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 += &#39;&#39;&#39;<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=&#39;button&#39;  value=&#39;删除&#39;/> </td>
                </tr>
                &#39;&#39;&#39; % row
    print(datas)

    # 把查询的数据，转换成动态内容
    with open(&#39;template/index.html&#39;, &#39;r&#39;, encoding=&#39;utf-8&#39;) as f:
        response_body = f.read()

    response_body = response_body.replace(&#39;{%datas%}&#39;, datas)

    response_first_line = &#39;HTTP/1.1 200 OK\r\n&#39;
    response_header = &#39;Server: Laoxiao_Server\r\n&#39;

    response = (response_first_line + response_header + &#39;\r\n&#39; + response_body).encode(&#39;utf-8&#39;)
    return response


# 处理没有找到对应的动态资源
def page_not_found():
    with open(&#39;static/404.html&#39;, &#39;rb&#39;) as f:
        response_body = f.read()  # 响应的主体页面内容(字节)
    # 响应头 （字符数据）
    response_first_line = &#39;HTTP/1.1 404 Not Found\r\n&#39;
    response_header = &#39;Server: Laoxiao_Server\r\n&#39;
    response = (response_first_line + response_header + &#39;\r\n&#39;).encode(&#39;utf-8&#39;) + response_body
    return response

2. クエリされたデータに基づいて動的コンテンツを取得します

以上がPython を使用してカスタム Web フレームワークを開発する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。