当我们使用get的方法传递参数的时候
params = {'key1': 'value1', 'key2': 'value2'}
async with session.get('http://httpbin.org/get',
params=params) as resp:
expect = 'http://httpbin.org/get?key1=value1&key2=value2'
assert str(resp.url) == expect
但是当我们使用post的方式传输的时候会有几种不同的情况,分别是表单的方式 json的方式 multioart的方式
前面两种对应postman中Body部分的form-data 和raw,因此在接收的时候需要区分
使用表单的方式接收的时候
data = await request.post()
使用raw的接收方式
data = await request.json()
import requests
import json
def requests_form():
url = 'http://httpbin.org/post'
data = {'k1':'v1', 'k2':'v2'}
response = requests.post(url, data)
return response
def requests_json():
url = 'http://httpbin.org/post'
data = s = json.dumps({'k1': 'v1', 'k2': 'v2'})
response = requests.post(url, data)
return response
def requests_multipart():
url = 'http://httpbin.org/post'
files = {'file': open('requests.txt', 'rb')} # requests.txt中包含一句“Hey requests”
response = requests.post(url, files=files)
return response
if __name__ == "__main__":
response1 = requests_form()
response2 = requests_json()
response3 = requests_multipart()
print("From形式提交POST请求:")
print(response1.text)
print("Json形式提交POST请求:")
print(response2.text)
print("Multipart形式提交POST请求:")
print(response3.text)
使用aiohttp搭建服务的时候,首先需要设置服务端:
# SET BASE_DIR
import os
import sys
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)
from aiohttp import web
import asyncio
from utils.utils import replace_punctuation
from src.main_client import retrieve_question_bank
import json
import re
import time
import hashlib
from loguru import logger
async def handle(request):
# 0 安全验证机制
logger.info('request start...', )
authorize_succ = auth(request)
if authorize_succ:
logger.info('request authorize success')
# 1 解析接收的数据
data = await request.json()
logger.info('get request json data quest_title:'+data.get('quest_title'))
quest_title = data['quest_title']
response = run_func(quest_title)
reply = json.dumps(response , ensure_ascii=False, indent=4)
logger.info('request end\n')
return web.Response(text=reply)
async def gethandle(request):
user_text = request.query['query']
reply = json.dumps(user_text , ensure_ascii=False, indent=4)
logger.info('request end\n')
return web.Response(text=reply)
def auth(request):
try:
#验证机制可以使用给服务设一个服务名 以及与接口调用方共同约定一个密码 然后加一个时间戳
#通过sha256 或者md5进行验证即可
request_timestamp = request.headers['timestamp']
client_encry = request.headers['encrysha256'] # 系统收到的sha256验证码
receive_params =servername + ',' + server_passwd + ',' + str(request_timestamp)
server_sha256 = hashlib.sha256(receive_params.encode('utf-8')).hexdigest() # 系统重新计算的sha256验证码
# 当前时间与请求发送时间的间隔需要在30min内,超时则验证失败
time_consume = time.time() - float(request_timestamp)
if int(time_consume) > 30 * 60:
print('请求超时!')
if client_encry != server_sha256:
print('sha256 验证失败!')
if client_encry == server_sha256 and time_consume <= 30 * 60:
return True
else:
return False
except:
logger.error('decode request info for authorize failed')
return False
async def init_app():
logger.add('runtime_{time}.log', rotation='00:00')
app = web.Application()
app.router.add_post('/post/', handle)
app.router.add_get('/get/', gethandle)
return app
if __name__ == '__main__':
loop = asyncio.get_event_loop()
app = loop.run_until_complete(init_app())
web.run_app(app, port='9086')
这里设置端口为9086
验证的机制为通过拼接 服务名 服务密码 时间戳
接口调用方会发送一个时间戳及调用方计算的验证码
我们通过接收的时间戳加上其他内容重新计算验证码即可 计算的方式可以使用sha256 md5等其他加密方法结可
注意 这部分数据可以直接放到header中进行传输
如下 header中的设置 做为字典格式进行传输即可 这里需要注意 headers中的参数名不要包含下划线'_',如果该服务在linux中运行的时候 ngix转发的时候会将下划线丢弃 导致参数无法解析的bug
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import aiohttp
import asyncio
import requests
import hashlib
import json
import time
async def main():
async with aiohttp.ClientSession() as session:
quest_title = "辛亥革命影响?"
server_name = 'youself'
server_password = 'youself'
timestamp = time.time()
encryption_key = server_name + ','+server_password + ','+str(timestamp)
encry_sha256 = hashlib.sha256(encryption_key.encode('utf-8')).hexdigest()
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'timestamp': str(timestamp),
'encrysha256': encry_sha256
}
# 若输入的问题文本为空 则非法输入
if quest_title or quest_content or quest_img_content:
payload = {'quest_title': quest_title}
payload = json.dumps(payload)
# dev 服务器
async with session.post('http://172.24.159.105:9086/post/', data=payload, headers=headers) as resp:
print('本次查询结果为:')
print(await resp.text())
else:
#使用get
params = {'query': quest_text }
async with session.get('http://172.24.227.247:9086/get/', params=params) as resp:
print('本次查询结果为:')
print(await resp.text())
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())