aiohttp和fastapi

fastapi一些小心得

关于请求体

path路径参数
query查询参数
这两个最先学的，都会出现在url路径里，类似
http://127.0.0.1:8000/user/555/items/1233?quote=It is good&auto_quote=false

这个时候发现，quote作为参数不太适合，更适合放在payload中。
例如server里使用Body()或者Field()模型属性。两种方法在server中的写法一样

@app.put("/items/{item_id}")
async def update_item(
    item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

cilent中

my_header = {"Content-Type": "application/json"}
item_id = "123"

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

class User(BaseModel):
    username: str
    full_name: str | None = None

async def main():
    async with aiohttp.ClientSession() as session:
        resp = await session.put(
            url=f"http://127.0.0.1:8000/items/{item_id}",
            headers=my_header,
            data=json.dumps(
                {
                    "item": Item(name="sss", price=10).model_dump(),
                    "user": User(username="xi").model_dump(),
                    "importance": 10,
                }
            ),
        )
        print(await resp.text())

asyncio.run(main=main())

关于传参

有些方面还是自己不懂请求的基础操作，如刚开始漏掉了Content-Type
传表单，这样可以

my_header = {'Content-Type':'application/x-www-form-urlencoded'}

async def main():
    async with aiohttp.ClientSession() as session:
        resp = await session.post(url=f"http://127.0.0.1:8000/login/",headers=my_header,data={"username":"123","password":"456"})
        print(await resp.text())

asyncio.run(main=main())

传json，这样可以
要注释解码器？

my_header={"Content-Type": "application/json"}
resp = await session.post(url=f"http://127.0.0.1:8000/apply_state/{name}",json=pwdModel(pwdA="123456").model_dump())

my_header={"Content-Type": "application/json"}

async def main():
    async with aiohttp.ClientSession() as session:
        resp = await session.post(url=f"http://127.0.0.1:8000/apply_state/EpicMo",headers=my_header,data=pwdModel(pwdA="123456").model_dump_json())
        print(await resp.text())

asyncio.run(main=main())

文件

首先补充一下aiohttp中的stream api

Streaming API — aiohttp 3.8.5 documentation
Streaming Response Content
While methods read(), json() and text() are very convenient you should use them carefully. All these methods load the whole response in memory. For example if you want to download several gigabyte sized files, these methods will load all the data in memory. Instead you can use the content attribute. It is an instance of the aiohttp.StreamReader class. The gzip and deflate transfer-encodings are automatically decoded for you:

async with session.get('https://api.github.com/events') as resp:
    await resp.content.read(10)
    ```
In general, however, you should use a pattern like this to save what is being streamed to a file:

```py
with open(filename, 'wb') as fd:
    async for chunk in resp.content.iter_chunked(chunk_size):
        fd.write(chunk)
It is not possible to use read(), json() and text() after explicit reading from content.

可能将python原生的file改为aiofiles也行，但笔者没有尝试

不包含文件时，表单数据一般用 application/x-www-form-urlencoded「媒体类型」编码。

但表单包含文件时，编码为 multipart/form-data。使用了 File，FastAPI 就知道要从请求体的正确位置获取文件。

编码和表单字段详见 MDN Web 文档的 POST 小节。

上传文件

一般使用multipart/form-data
aiohttp.FormData()上传时会自动生成boundary

my_data = aiohttp.FormData()
my_data.add_field('file',open('test.py',"rb"))
my_data.add_field("fileb",open('test222.py',"rb"))
my_data.add_field("token","45s4df545sdaf")

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.post(url=requests_url,data=my_data) as resp:
            print(resp.status)
            print(await resp.text())

asyncio.run(main=main())

# 接受大文件
async def file_sender(file_name=None):
    async with aiofiles.open(file_name, 'rb') as f:
        chunk = await f.read(64*1024)
        while chunk:
            yield chunk
            chunk = await f.read(64*1024)

# Then you can use file_sender as a data provider:

async with session.post('http://httpbin.org/post',
                        data=file_sender(file_name='huge_file')) as resp:
    print(await resp.text())