Client

ClientSession

response

  • resp.status
一次性读取内容

  • resp.text()

    • 文本 str 类型
    • resp.text(encoding='gbk')

  • resp.read()

    • bytes 类型

  • resp.json()

    • 提取 json
流模式读取 Streaming Response Content
  • resp.content.read(10)

request

  • session.get
  • session.put
关键字参数 arguments
params
json
  • session.post(url, json={'test': 'value'})

Post 请求

sesson.post https://docs.aiohttp.org/en/stable/client_quickstart.html#more-complicated-post-requests

传输 dict

  • 用途：用于传输表格数据 form-encoded data, like html form

    1
2

    payload = {'name': 'lily', 'score': 80}
session.post(url, data=payload)
传输 bytes

This data will be posted directly and content-type set to ‘application/octet-stream’

1

session.post(url, data=b'this is just one line.')
传输 json
1
2

async with session.post(url, json={'a': 1}) as resp:
pass
传输文件 file
  • 使用 data 关键字参数
  • 单个文件模式
  • 多个文件模式
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# * 单个文件
url = 'http://httpbin.org/post'
files = {'file': open('report.xls', 'rb')}

await session.post(url, data=files)

# * 多个文件
url = 'http://httpbin.org/post'
data = FormData()
data.add_field('file',
               open('report.xls', 'rb'),
               filename='report.xls',
               content_type='application/vnd.ms-excel')

await session.post(url, data=data)

# * 流式传输
with open('massive-body', 'rb') as f:
    await session.post('http://httpbin.org/post', data=f)

WebSockets

创建

  • session.ws_connect(url)

    • 返回值：ClientWebSocketResponse

      • 类似 Response like
 1
 2
 3
 4
 5
 6
 7
 8
 9
10

async with session.ws_connect('http://example.org/ws') as ws:
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            if msg.data == 'close cmd':
                await ws.close()
                break
            else:
                await ws.send_str(msg.data + '/answer')
            elif msg.type == aiohttp.WSMsgType.ERROR:
                break

接受信息

1
2
3
4
5
6

# * 法一
await ws.receive()

# * 法二
async for msg in ws:
    pass

发送信息

1

ws.send_str('data')

Timeouts 超时

https://docs.aiohttp.org/en/stable/client_quickstart.html#timeouts

指定方法

  • 创建 session 时，ClientSession(timeout=timeout)

    • timeout = aiohttp.ClientTimeout(total=60)
  • 请求时，session.get(url, timeout=timeout)

Server

接口

总体格式

  • 要素

    • web 应用

      • web.Application() 类

        • 路径处理工具 handler

  • 注意接口

    • 接受一个 Request

      • async def hello(request)
    • 返回一个 web.Response()
    • 运行 web 应用
  • web.run_app
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import asyncio
from aiohttp import web

# * handle, Request 处理工具
async def hello(request):
    return web.Response(text='Hello World!')


def main():
    # * 声明web 应用
    app = web.Application()

# * 添加路径处理工具
app.add_routes([web.get('/', hello)])

# * 运行web 应用
web.run_app(app)

if __name__ == '__main__':
    main()

handler 定义与添加

使用 web.get 设置（Django 风格）
  • 先定义
  • 再逐个手动添加
1
2
3
4
5
6
7

# 1) 定义
async def hello(request):
    return web.Response(text='Hello World!')

# 2) 添加到 routes 列表
routes = [web.get('/', hello)]
app.add_routes(routes)
使用装饰器（flask 风格）

  • 定义时自动添加

    • 定义和添加放到一起

  • 装饰器制作工具

    • web.RouteTableDef()
1
2
3
4
5
6
7
8
9

routes = web.RouteTableDef()

@routes.get('/')
async def hello(request):
    return web.Response(text="Hello, world")

app = web.Application()
app.add_routes(routes)
web.run_app(app)