aiohttp 服务端与客户端的使用注意事项

当我们使用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())

标签：aiohttp,request,json,quest,print,import,requests,服务端,客户端
来源： https://blog.csdn.net/u014028063/article/details/114657419