【FastAPI】异步+网络请求

• 前言：当我们碰到IO阻塞的时，使用fastapi异步框架，需要配合异步模块使用
• 1.网络请求，httpx

import httpx
import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request

app = FastAPI()

@app.get("/api")
async def api(request: Request):
    data_dict = await request.json()
    print(data_dict)

    async with httpx.AsyncClient(proxies={}, cookies={}, headers={}) as client:
        res = await client.get(
            url="https://movie.douban.com/j/search_subjects?type=movie&tag=%E7%83%AD%E9%97%A8&sort=recommend&page_limit=20&page_start=20",
            params={},
            cookies={},
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36"
            }
        )
        douban_dict = res.json()

    print(douban_dict)
    return {"data": "ok"}


if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=9001)

• 2.数据库相关，注：一般不需要，正常查询速度够用，即使慢也是因为慢SQL导致
  • 2.1：Redis，pip3 install aioredis

import httpx
import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request
import aioredis

app = FastAPI()

@app.get("/api")
async def api(request: Request):
    data_dict = await request.json()
    print(data_dict)

    #conn = await aioredis.Redis(host="127.0.0.1", port=6379, password="123456")
    #await conn.set("k1", "v1")
    #value = await conn.get("k1")
    #print(value)
    #await conn.close()

    async with aioredis.Redis(host="127.0.0.1", port=6379, password="123456") as conn:
        await conn.set("k1", "v1")
        value = await conn.get("k1")
        print(value)

    return {"data": "ok"}

if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=9001)

• 2.2：mysql，pip install aiomysql

import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request
import aiomysql

app = FastAPI()

@app.get("/api")
async def api(request: Request):
    data_dict = await request.json()
    print(data_dict)

    async with aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123456', db='mysql') as conn:
        cur = await conn.cursor()
        await cur.execute("SELECT Host,User FROM user")
        result = await cur.fetchall()
        print(result)

    return {"data": "ok"}


if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=9001)