Python 基于http.server模块实现简单http服务的代码举例

# -*- coding:utf-8 -*-
import json
import traceback
import uuid
from http.server import HTTPServer, ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
class MyHTTPRequestHandler(BaseHTTPRequestHandler):
    def _send_response(self, status_code, content_type, body):
        '''发送响应信息'''
        self.send_response(status_code)
        self.send_header('Content-type', content_type)
        self.send_header('Cache-Control', 'no-store')  # 防止缓存旧编码响应
        self.end_headers()
        self.wfile.write(body.encode('utf-8'))
    def _handle_home(self):
        '''访问主页请求处理'''
        html = '<meta http-equiv="Content-Type" content="text/html; charset=utf-8"><h1>Home Page</h1>'
        self._send_response(200, 'text/html; charset=utf-8', html)
    def _handle_404(self):
        '''请求不存在资源处理''China编程'
        # json_respose = {"success": False, "message": "Not Found"}
        # self._send_response(404, 'application/json;charset=utf-8', json.dumps(json_respose))
        # Content-Type 指定 charset=utf-8 可避免浏览器GET请求，界面中文显示乱码问题
        self._send_response(404, 'text/html; charset=utf-8', "<h1>404 Not Found</h1>")
    def _handle_login(self, login_req_data):
        '''处理登录请求'''
        try:
            data = json.loads(login_req_data)
            username = data.get('username')
            password = data.get('password')
            ip = data.get('ip')
            response = {
                'code': 0,
                'token': uuid.uuid4().hex,
                'description': 'success'
            }
            self._send_response(200, 'application/json; charset=utf-8', json.dumps(response))
        except Exception as e:
            error_msg = traceback.format_exc()
            print(error_msg)
            response = {
                'code': 1,
                'token': '',
                'description': error_msg
            }
            self._send_response(500, 'application/json; charset=utf-8', json.dumps(response))
    def do_GET(self):
        '''处理GET请求'''
        parsed_path = urlparse(self.path)
        path = parsed_path.path
        query_params = parse_qs(parsed_path.query)  # 获取URL携带的查询参数
        # print('收到GET请求参数:', query_params)
        if path == '/':
            self._handle_home()
        else:
            self._handle_404()
    def do_POST(self):
        '''处理POST请求'''
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length)
        parsed_path = urlparse(self.path)
        path = parsed_path.path
        query_params = parse_qs(parsed_path.query)  # 获取URL 查询参数
        # print("收到POST数据:", post_data.decode())
        # 路由匹配逻辑
        if path == '/':
            self._handle_home()
        elif path == '/north/login':
            self._handle_login(post_data.decode())
        else:
            self._handle_404() 
if __name__ == '__main__':
    # server = HTTPServer(('0.0.0.0', 8000), MyHandler) # 阻塞式运行
    server = ThreadingHTTPServer(('localhost', 8000), MyHTTPRpythonequestHandler)
    print('正在启动服务，访问地址：http://localhost:8000')
    server.serve_forever()

class Subsystem():
    def __init__(self, http_server_port):
        self.http_server_port = http_server_port
        self.server = ThreadingHTTPServer(('0.0.0.0', self.http_server_port), lambda *args: MyHTTPRequestHandler(self, *args))
        self.north_server_info = {}
    def start_http_server(self):
        self.server.serve_forever()
    def push_data(self, data):
        '''测试函数'''
        logger.info(f'pushing data to server')
    def update_north_server_info(self, data):
        '''测试函数'''
        self.north_server_info[data.get('ip')] = data
class MyHTTPRequestHandler(BaseHTTPRequestHandler):
    def __init__(self, subsystem, *args, **kwargs):
        self.subsystem = subsystem  # 保存SubSystem实例引用
        super().__init__(*args, **kwargs)
    # ....略
    def _handle_login(self, login_req_data):
        '''处理登录请求'''
        try:
            data = json.loads(login_req_data)
            username = data.get('username')
            password = data.get('password')
            ip = data.get('ip')
            response = {
                'code': 0,
                'token': uuid.uuid4().hex,
                'description': 'success'
            }
            self.subsystem.push_data('')
            self.subsystem.update_north_server_info({'username': username, 'password': password, 'ip': ip})
            self._send_response(200, 'application/json; charset=utf-8', json.dumps(response))
        except Exception as e:
            error_msg = traceback.format_exc()
            logger.error(error_msg)
            response = {
                'code': 1,
                'token':'',
                'description': error_msg
            }
            self._send_response(500, 'application/json; charset=utf-8', json.dumps(response))
# 测试
if __name__ == '__main__':
    http_port = 8000
    for i in range(2):
        system = Subsystem(http_port)
        thread = threading.Thread(target=system.start_http_server)
        thread.start()
        http_port += 1

def run(server_class=HTTPServer, handler_class=BaseHTTPRequestHandler):
    server_address = ('', 8000)
    httpd = server_class(server_address, handler_class)
    httpd.serve_forever()