WSGI（Web Server Gateway Interface，web服务器网关接口）主要规定了服务器端和应用程序之间的接口，即规定了请求的URL到后台处理函数之间的映射该如何实现。wsgiref是一个帮助开发者开发测试的Python内置库，程序员可以通过这个库了解WSGI的基本运行原理，但是不能把它用在生产环境上。
WSGI处理过程
- 浏览器到wsgi server :浏览器发送的请求会先到wsgi server
- environ: wsgi server 会将http请求中的参数等信息封装到environ(一个字典) 中
- wsgi server 到wsgi app :app就是我们编写的后台程序，每个url会映射到对于的入口处理函数，wsig server调用后台app时，会将environ和wsgi server中自己一个start_response函数注入到后台app中
- 逻辑处理：后台函数需要接受environ和start_response，进行逻辑处理后返回一个可迭代对象，可迭代对下中的元素为http正文
- wsgi app 到wsgi server:后台函数处理完后，会先调用start_response函数将http状态码，报文头等信息(响应头)返回给wsgi server，然后再将函数的返回值作为http正文(响应body)返回给wsgi server.
- wsgi server 到浏览器：wsgi server 将从app中等到的所有信息封装作为一个response返回给浏览器
import hashlib,requests,json,time,urllib.parse
from http import HTTPStatus
import dashscope
from wsgiref.simple_server import make_server
errStr ='''
{
"code" : -1,
"msg" : "not support"
}
'''
notStr = '''
{
"code" : -2,
"msg" : "not allowed"
}
'''
# For prerequisites running the following sample, visit https://help.aliyun.com/document_detail/611472.html
dashscope.api_key="api_key" # 自己的api_key
def al(cont):
messages = [{'role': 'system', 'content': 'You are a helpful assistant.'},
{'role': 'user', 'content': '%s'%cont}]
response = dashscope.Generation.call(
dashscope.Generation.Models.qwen_turbo,
messages=messages,
result_format='message', # set the result to be "message" format.
)
if response.status_code == HTTPStatus.OK:
content = response['output']['choices'][0]['message']['content']
# print(content)
return content
else:
print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
response.request_id, response.status_code,
response.code, response.message
))
return errStr
# 通义千问
def Tongyi(cont):
url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
payload = json.dumps({
"model": "qwen-turbo",
"input": {
"messages": [
{
"role": "system",
"content": "You are a helpful assistant."
},
{
"role": "user",
"content": cont
}
]
},
"parameters": {}
})
headers = {
'Content-Type': 'application/json',
'Authorization': 'key' # 换成自己的key
}
response = requests.request("POST", url, headers=headers, data=payload)
msg = response.json()['output']['text']
return msg
# 青云客
def qingyunk(cont):
url = "http://api.qingyunke.com/api.php?key=free&appid=0&msg=%s" % (urllib.parse.quote(cont))
html = requests.get(url)
msg = html.json()['content']
return msg
def RunServer(environ,start_response):
# 添加回复内容的http头部信息，支持多个
headers = {'Content-Type': 'application/json', 'Custom-head1': 'Custom-info1'}
# environ 包含当前环境信息与请求信息，为字符串类型的键值对
current_url = environ['PATH_INFO']
current_content_length = environ['CONTENT_LENGTH']
current_request_method = environ['REQUEST_METHOD']
# 获取body json 内容转换为python对象
current_req_body = environ['wsgi.input'].read(int(environ['CONTENT_LENGTH']))
current_req_json = json.loads(current_req_body)
# 获取输入值
cont = current_req_json['cont']
cont = urllib.parse.unquote(cont)
print(cont)
#打印请求信息
print("REQUEST METHOD:",current_request_method)
print("REQUEST URL:",current_url)
print("REQUEST BODY:",current_req_json)
#根据不同url回复不同内容
if current_url == "/qingyunk":
if current_request_method == "GET":
result = Tongyi(cont)
print(result)
# 拼装回复报文
successStr = '''
{
"code":200,
"msg":"success",
"data":{
"content":"%s"
}
}
''' % (result)
start_response("200 OK", list(headers.items()))
return [successStr.encode("utf-8"), ]
else:
start_response('403 not allowed',list(headers.items()))
return [notStr.encode("utf-8"),]
elif current_url == "/Tongyi":
result = qingyunk(cont)
print(result)
# 拼装回复报文
successStr = '''
{
"code":200,
"msg":"success",
"data":{
"content":"%s"
}
}
''' % (result)
start_response("200 OK", list(headers.items()))
return [successStr.encode("utf-8"), ]
elif current_url == '/al':
result = al(cont)
# 拼装回复报文
successStr = '''
{
"code":200,
"msg":"success",
"data":{
"content":"%s"
}
}
''' % (result)
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__":
httpd = make_server('', 10000, RunServer)
host, port = httpd.socket.getsockname()
print('Serving running', host, 'port', port)
httpd.serve_forever()
WSGI web服务器
- 本质上是一个TCP服务器，监听在特定的端口上。
- 支持HTTP协议，能够解析HTTP请求报文，能够按HTTP协议将响应数据封装为报文并返回给浏览器。
- 实现了WSGI协议，该协议约定了和应用程序之间的接口，即url到app之间的映射。
WSGI应用程序
- 遵从WSGI协议。
- 本身是一个可调用对象。
- 调用start_response，返回响应头部。
- 返回包含正文的可迭代对象。