文章目录

前言

项目经常需要HTTP对接，模拟HTTP client请求可以使用postman测试，模拟HTTP server回复该如何处理？本文介绍通过python搭建HTTP服务器的过程。

先装python，然后装PyCharm。

python安装、下载说明，看这部分
https://www.runoob.com/python3/python3-install.html

PyCharm安装、下载说明，看这部分
PyCharm 下载地址 : https://www.jetbrains.com/pycharm/download/
PyCharm 安装地址：http://www.runoob.com/w3cnote/pycharm-windows-install.html

一、HTTP服务器搭建

如下示例搭建了一个HTTP服务，可根据URL、HTTP请求方法、请求的内容（JSON）来回复不同的内容，可根据实际项目修改如下脚本即可。 
#!/usr/bin/python

from wsgiref.simple_server import make_server
import json

#输出回复时，通过string.encode()指定输出的文字编码方式，string.encode('gb2312')、string.encode('utf-8')、string.encode('gbk')
connStr ='''
{ 
	"code" : 0, 
	"msg" : "SUCCESS 成功"
}
'''

heartbeatStr ='''
{ 
	"code" : 0, 
	"msg" : "SUCCESS 成功"
}
'''

heartbeatErrStr ='''
{ 
	"code" : -1, 
	"msg" : "FAILED 无效"
}
'''

infoGetStr ='''
{ 
	"code" : 0, 
	"msg" : "GET 成功"
}
'''

infoPostStr ='''
{ 
	"code" : 0, 
	"msg" : "POST 成功"
}
'''

errStr ='''
{ 
	"code" : -1, 
	"msg" : "not support"
}
'''

def RunServer(environ, start_response):

    #添加回复内容的HTTP头部信息，支持多个
    headers = {'Content-Type': 'application/json', 'Custom-head1': 'Custom-info1'}

    # environ 包含当前环境信息与请求信息，为字符串类型的键值对
    current_url = environ['PATH_INFO']
    current_content_type = environ['CONTENT_TYPE']
    current_content_length = environ['CONTENT_LENGTH']
    current_request_method = environ['REQUEST_METHOD']
    current_remote_address = environ['REMOTE_ADDR']
    current_encode_type = environ['PYTHONIOENCODING']        #获取当前文字编码格式，默认为UTF-8
    '''
    HTTP客户端请求的其他头部信息（Host、Connection、Accept等），对应environ内容为“HTTP_XXX”，
    例如：请求头部为"custom-header: value1",想获取custom-header的值使用如下方式：
    '''
    #current_custom_header = environ['HTTP_CUSTOM_HEADER']

    #获取 body JSON内容转换为python对象
    current_req_body = environ['wsgi.input'].read(int(environ['CONTENT_LENGTH']))
    current_req_json = json.loads(current_req_body)

    #打印请求信息
    #print("environ:", environ)
    print("REQUEST remote ip:", current_remote_address)
    print("REQUEST method:", current_request_method)
    print("REQUEST URL:", current_url)
    print("REQUEST Content-Type:", current_content_type)
    #print("REQUEST Custom-header:", current_custom_header)
    print("REQUEST body:", current_req_json)

    #根据不同URL回复不同内容
    if current_url == "/connect":
        start_response("200 OK", list(headers.items()))
        return [connStr.encode("utf-8"), ]

    elif current_url == "/keepalive":
        # 根据 body JSON内容，回复不同内容
        '''
        判断JSON请求的内容是否合法,假设请求内容如下：判断 cmd 等于 keepalive, interval 大于30
        {
            "cmd": "keepalive",
            "data":
            {
                "interval": 30
            }
        }
        '''
        if current_req_json['cmd'] == 'keepalive':
            if current_req_json['data']['interval'] >= 30:
                start_response("200 OK", list(headers.items()))
                return [heartbeatStr.encode("utf-8"), ]
            else:
                start_response("410 err", list(headers.items()))
                return [heartbeatErrStr.encode("utf-8"), ]
        else:
            start_response("410 err", list(headers.items()))
            return [heartbeatErrStr.encode("utf-8"), ]

    elif current_url == "/info":
        # 根据不同请求方法回复不同内容
        if current_request_method == 'GET':
            start_response("200 OK", list(headers.items()))
            return [infoGetStr.encode("utf-8"), ]

        elif current_request_method == 'POST':
            start_response("200 OK", list(headers.items()))
            return [infoPostStr.encode("utf-8"), ]

        else:
            start_response("404 not found", list(headers.items()))
            return [errStr.encode("utf-8"), ]

    else:
        start_response("404 not found", list(headers.items()))
        return [errStr.encode("utf-8"), ]

if __name__ == "__main__":
    #10000为HTTP服务监听端口，自行修改
    httpd = make_server('', 10000, RunServer)
    host, port = httpd.socket.getsockname()
    print('Serving running', host, 'port', port)
    httpd.serve_forever()

二、功能强化

部分对接要求回复时返回的参数为请求的部分参数，即请求是带了A参数，要求回复时原样返回A参数，且A一直在变化，比如事件ID、时间戳等等。针对以上问题，为了方便调试，如下代码可根据请求的参数动态回复内容。

举例：

eventId、deviceIndexCode、idCardNo、name为请求时的JSON报文入参，要求返回报文时原样返回。
请求报文postman截图如下：
在这里插入图片描述

#!/usr/bin/python

from wsgiref.simple_server import make_server
import json

#输出回复时，通过string.encode()指定输出的文字编码方式，string.encode('gb2312')、string.encode('utf-8')、string.encode('gbk')
eventId = ''
deviceIndexCode = ''
idCardNo = ''
personName = ''

errStr ='''
{ 
	"code" : -1, 
	"msg" : "not support"
}
'''
def RunServer(environ, start_response):

    #添加回复内容的HTTP头部信息，支持多个
    headers = {'Content-Type': 'application/json', 'Custom-head1': 'Custom-info1'}

    # environ 包含当前环境信息与请求信息，为字符串类型的键值对
    current_url = environ['PATH_INFO']
    current_content_type = environ['CONTENT_TYPE']
    current_content_length = environ['CONTENT_LENGTH']
    current_request_method = environ['REQUEST_METHOD']
    current_remote_address = environ['REMOTE_ADDR']
    current_encode_type = environ['PYTHONIOENCODING']        #获取当前文字编码格式，默认为UTF-8

    #获取 body JSON内容转换为python对象
    current_req_body = environ['wsgi.input'].read(int(environ['CONTENT_LENGTH']))
    current_req_json = json.loads(current_req_body)

    #打印请求信息
    #print("environ:", environ)
    print("REQUEST remote ip:", current_remote_address)
    print("REQUEST method:", current_request_method)
    print("REQUEST URL:", current_url)
    print("REQUEST Content-Type:", current_content_type)
    print("REQUEST body:", current_req_json)

    eventId = current_req_json['eventId']
    deviceIndexCode = current_req_json['deviceIndexCode']
    idCardNo = current_req_json['idCardNo']
    personName = current_req_json['name']

    #拼装回复报文
    successStr = '''
    {
        "code":0,"msg":"success",
        "data":{
            "eventId":"%s",
            "deviceIndexCode":"%s",
            "idCardNo":"%s",
            "personName":"%s"
        }
    }
    ''' % (eventId, deviceIndexCode, idCardNo, personName)

    #根据不同URL回复不同内容
    if current_url == "/check":
        start_response("200 OK", list(headers.items()))
        return [successStr.encode("utf-8"), ]

    else:
        start_response("404 not found", list(headers.items()))
        return [errStr.encode("utf-8"), ]

if __name__ == "__main__":
    #10000为HTTP服务监听端口，自行修改
    httpd = make_server('', 10000, RunServer)
    host, port = httpd.socket.getsockname()
    print('Serving running', host, 'port', port)
    httpd.serve_forever()

最后效果如下图：
在这里插入图片描述

# 网络 # http # java
Logo
华为开发者空间

华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。

更多推荐

cover

解析openGauss账本数据库

avatar 华为开发者空间
cover

GaussDB高校行走进哈尔滨，与两校学子共话数据库人才发展

avatar 华为开发者空间
cover

GaussDB数据库高可用部署方案

avatar 华为开发者空间