Python 基于http.server模塊實現(xiàn)簡單http服務(wù)
測試環(huán)境
win11專業(yè)版
python 3.9
代碼實現(xiàn)
# -*- 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):
'''發(fā)送響應(yīng)信息'''
self.send_response(status_code)
self.send_header('Content-type', content_type)
self.send_header('Cache-Control', 'no-store') # 防止緩存舊編碼響應(yīng)
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):
'''請求不存在資源處理'''
# 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攜帶的查詢參數(shù)
# print('收到GET請求參數(shù):', 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 查詢參數(shù)
# print("收到POST數(shù)據(jù):", 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), MyHTTPRequestHandler)
print('正在啟動服務(wù)，訪問地址：http://localhost:8000')
server.serve_forever()
擴展：如果我們希望服務(wù)在處理請求的時，調(diào)用其它類實例的方法，或者更新其它類實例的屬性，咋處理呢？
答案：將其它類實例初始化為RequestHandler的類屬性，然后在相關(guān)請求處理函數(shù)中進行調(diào)用
示例
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):
'''測試函數(shù)'''
logger.info(f'pushing data to server')
def update_north_server_info(self, data):
'''測試函數(shù)'''
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
相關(guān)介紹
模塊簡介
http.server 模塊定義了用于實現(xiàn)HTTP服務(wù)器（Web服務(wù)器）的類。
其中一個類，HTTPServer是socketserver.TCPServer子類。它創(chuàng)建并監(jiān)聽HTTP套接字，將請求分派給處理程序。創(chuàng)建和運行服務(wù)器的代碼示例如下：
def run(server_class=HTTPServer, handler_class=BaseHTTPRequestHandler):
server_address = ('', 8000)
httpd = server_class(server_address, handler_class)
httpd.serve_forever()
類及相關(guān)函數(shù)簡介
-
class http.server.HTTPServer(server_address, RequestHandlerClass)該類基于基于
TCPServer類構(gòu)建，通過將將server地址存儲為名為server_name和server_port的實例變量。通過handler訪問服務(wù)，通常是通過handler的server實例變量 -
class http.server.ThreadingHTTPServer(server_address, RequestHandlerClass)同
HTTPServer一樣，除了通過使用ThreadingMixIn使用線程處理請求。 3.7版本中新增
HTTPServer 和ThreadingHTTPServer 必須在實例化時給它一個RequestHandlerClass，此模塊提供了三種不同的變體：
-
class http.server.BaseHTTPRequestHandler(request, client_address, server)此類用于處理到達服務(wù)器的HTTP請求。就其本身而言，它無法響應(yīng)任何實際的HTTP請求；它必須被子類化以處理每個請求方法（例如
GET或POST）。BaseHTTPRequestHandler提供了許多類和實例變量以及子類使用的方法。BaseHTTPRequestHandler實例變量:-
client_address包含一個
(host, port)形式，表示客戶端地址的元組 -
server包含服務(wù)器實例
-
close_connectionhandle_one_request()返回前需要設(shè)置的布爾值。指示是否可能需要另一個請求，或者是否應(yīng)該關(guān)閉連接。 -
requestline包含HTTP請求行的字符串表示，不含終止
CRLF。此屬性應(yīng)通過[handle_one_request()設(shè)置。如果沒有處理有效的請求行，則應(yīng)將其設(shè)置為空字符串。 -
command包含命令(請求類型)。例如，
'GET'. -
path包含請求路徑。 如果URL的查詢參數(shù)部分存在，則
path包括查詢參數(shù)。 -
request_version包含請求版本字符串，形如'HTTP/1.0'`。
-
headers保存由
MessageClass類變量指定的類的實例。此實例解析和管理HTTP請求中的請求頭。parse_headers()函數(shù)來自http.client，用于解析標頭，它要求http請求提供有效的RFC 2822樣式請求頭。 -
rfile一個
io.BufferedIOBase輸入流，準備從可選輸入數(shù)據(jù)的開頭讀取。 -
wfile包含用于將響應(yīng)寫回客戶端的輸出流。在寫入此流時，必須正確遵守HTTP協(xié)議，以實現(xiàn)與HTTP客戶端的成功交互操作。在3.6版本中更改：這是一個
io.BufferedIOBase流。
BaseHTTPRequestHandler屬性:-
server_version指定服務(wù)器軟件版本。你可能希望覆蓋此內(nèi)容。格式是多個空格分隔的字符串，其中每個字符串的格式為
name[/version]。例如，'BaseHTTP/0.2'。 -
sys_version包含Python系統(tǒng)版本，其形式和
version_string方法及server_version]類變量使用的相同。例如，'Python/1.4'。 -
error_message_format指定
send_error()應(yīng)使用的格式字符串，用于構(gòu)建對客戶端的錯誤響應(yīng)的。默認情況下，基于傳遞給send_error的狀態(tài)代碼，用responds中的變量填充該字符串。 -
error_content_type指定發(fā)送到客戶端的錯誤響應(yīng)的
Content-TypeHTTP請求頭。默認值為'text/html'。 -
protocol_version這指定了響應(yīng)中使用的HTTP協(xié)議版本。如果設(shè)置為
'HTTP/1.1'，服務(wù)器將允許HTTP持久連接；但是，服務(wù)器在對客戶端的所有響應(yīng)中必須包含一個準確的Content-Length請求頭(使用send_header())。為了向后兼容，設(shè)置默認為“HTTP/1.0”。 -
MessageClass指定一個
email.message.Message之類的類，用于解析HTTP頭。通常，這是不可覆蓋的，默認為http.client.HTTPMessage。 -
responses此屬性包含整數(shù)錯誤代碼到包含短消息和長消息的兩個元素元組的映射。例如，
{code: (shortmessage, longmessage)}。*shortmessage*通常用作錯誤響應(yīng)中的message鍵，longmessage用作explain鍵。供send_response_only和send_error()方法使用。
BaseHTTPRequestHandler實例方法:-
handle()調(diào)用
handle_one_request()實現(xiàn)適當?shù)?code>do_*() 方法。 -
handle_one_request()此方法將解析請求并將其分發(fā)到適當?shù)?code>do_*()方法。你應(yīng)永遠不需要重寫它。
-
handle_expect_100()當符合HTTP/1.1標準的服務(wù)器收到
Expect: 100-continue請求頭時，它會以100 Continue和“200 OK頭作為響應(yīng)。如果服務(wù)器不希望客戶端繼續(xù)，則可以重寫此方法以引發(fā)錯誤。例如，服務(wù)器可以選擇發(fā)送417 Expectation Failed作為響應(yīng)頭，并返回False。3.2版本中的新功能
-
send_error(code, message=None, explain=None)向客戶端發(fā)送并記錄完整的錯誤回復(fù)。數(shù)字code指定HTTP錯誤代碼，message作為可選的、簡短的、人類可讀的錯誤描述。explain參數(shù)可用于提供有關(guān)錯誤的更詳細信息；它將使用
error_message_format屬性進行格式化，并在一組完整的標頭之后作為響應(yīng)體發(fā)出。response屬性包含message和explain的默認值，如果沒有提供值，將使用這些值；對于未知代碼，兩者的默認值都是字符串???。如果方法是HEAD或響應(yīng)代碼是以下代碼之一:1xx,204 No Content,205 Reset Content,304 Not Modified，則請求體將為空。在版本3.4中變更：錯誤響應(yīng)包含
Content-Length頭。添加了explain參數(shù)。 -
send_response(code, message=None)添加一個響應(yīng)頭添加到headers緩沖區(qū)，并記錄接受的請求。HTTP響應(yīng)行被寫入內(nèi)部緩沖區(qū)，后面跟隨Server和Date響應(yīng)頭。這兩個響應(yīng)頭的值分別從
version_string()和date_time_string()方法中獲取。如果服務(wù)器不打算使用send_header()發(fā)送任何其他請求頭，則send_response()后面應(yīng)該跟一個end_headers()調(diào)用。3.3版本中變更：header存儲在內(nèi)部緩沖區(qū)，且需要顯示調(diào)用
end_headers()。 -
send_header(keyword, value)將HTTP頭添加到內(nèi)部緩沖區(qū)，當
end_headers()或者flush_headers()被調(diào)用時，該緩沖區(qū)中的內(nèi)容將被寫入到輸出流。keyword應(yīng)指定header，value指定其值。請注意，在send_headers調(diào)用完成后，必須調(diào)用end_headers()才能完成操作。版本3.2中變更：HTTP頭存儲在內(nèi)部緩沖區(qū)中
-
send_response_only(code, message=None)僅發(fā)送響應(yīng)頭，服務(wù)器向客戶端發(fā)送
100 Continue響應(yīng)時使用。響應(yīng)頭未緩沖，直接發(fā)送到輸出流。如果未指定message，則發(fā)送與響應(yīng)code對應(yīng)的HTTP消息。版本3.2中的新功能
-
end_headers()在header緩沖區(qū)中添加一個空行（表示響應(yīng)中HTTP頭的結(jié)束）并調(diào)用
flush_headers()。版本3.2中變更：緩沖的HTTP頭被寫入輸出流
-
flush_headers()最后將HTTP頭發(fā)送到輸出流并刷新內(nèi)部HTTP頭緩沖區(qū)。3.3版本中的新功能。
-
log_request(code='-', size='-')記錄已接受（成功）的請求。code應(yīng)指定與響應(yīng)關(guān)聯(lián)的數(shù)字HTTP code。如果可獲取響應(yīng)的大小，則應(yīng)將其作為size參數(shù)傳遞。
-
log_error(...)當請求無法滿足時記錄錯誤。默認情況下，它將消息傳遞給
log_message()，因此它采用相同的參數(shù)（format和其它參數(shù)）。 -
log_message(format, ...)將任意消息記錄到
sys.stderr。這通常會被重寫以創(chuàng)建自定義錯誤日志記錄機制。format參數(shù)是一個標準的類printf風格的格式字符串，其中log_message()的附加參數(shù)作為格式化的輸入。客戶端ip地址和當前日期和時間都會作為記錄的每條消息的前綴。 -
version_string()返回服務(wù)器軟件的版本字符串。這是
server_version以及sys_version屬性的混合。 -
date_time_string(timestamp=None)返回timestamp給出的日期和時間（必須為
None或采用time.time()返回的格式），格式化為消息頭。如果省略timestamp，則使用當前日期和時間。結(jié)果看起來像'Sun, 06 Nov 1994 08:49:37 GMT'。 -
log_date_time_string()返回當前日期和時間，格式化為日志記錄格式。
-
address_string()返回客戶端地址。
3.3版本中變更：以前執(zhí)行了名稱查找。為了避免名稱解析延遲，它現(xiàn)在總是返回IP地址。
-
-
class http.server.SimpleHTTPRequestHandler(request, client_address, server, directory=None)介紹略
-
class http.server.``CGIHTTPRequestHandler(*request*, client_address, server)介紹略
參考鏈接
作者：授客
微信/QQ：1033553122
全國軟件測試QQ交流群：7156436
Git地址：https://gitee.com/ishouke
友情提示：限于時間倉促，文中可能存在錯誤，歡迎指正、評論！
作者五行缺錢，如果覺得文章對您有幫助，請掃描下邊的二維碼打賞作者，金額隨意，您的支持將是我繼續(xù)創(chuàng)作的源動力，打賞后如有任何疑問，請聯(lián)系我!!!
微信打賞
支付寶打賞 全國軟件測試交流QQ群
浙公網(wǎng)安備 33010602011771號