async视图必须配Uvicorn,flask run会静默退化为同步执行;需用uvicorn启动、禁用gunicorn/gevent;async视图中禁用requests,改用httpx.AsyncClient或aiohttp;SQLAlchemy须配置async_engine和AsyncSession;before_request不支持async,应改用ASGI中间件或视图内处理。

async视图必须配Uvicorn,flask run会静默退化
写了async def视图函数,但用flask run启动,它不会报错,也不会真正异步——而是降级为同步执行,await语句被当成普通调用,整个请求仍占一个线程,毫无并发收益。
根本原因是flask run基于Werkzeug(WSGI),不支持ASGI事件循环。你必须换服务器:
- 安装
uvicorn:pip install uvicorn - 启动命令必须是
uvicorn main:app(假设main.py里定义了app) - 禁用
gunicorn --worker-class sync或gevent——它们仍是WSGI模型,无法调度async协程
运行后检查日志:若看到Uvicorn running on http://...且无RuntimeWarning: Running asyncio.run() in a loop that's already running,说明环境就绪。
async视图里不能用requests,必须换aiohttp或httpx.AsyncClient
这是最常踩的坑:requests.get()是阻塞调用,一旦出现在async视图中,会卡死整个Uvicorn worker的事件循环,所有后续请求排队等待,表现就是“突然全接口变慢”或“503超时”。
立即学习“Python免费学习笔记(深入)”;
正确做法是使用原生异步HTTP客户端:
-
aiohttp(需手动管理session生命周期) -
httpx.AsyncClient(推荐,API接近requests,支持HTTP/2)
示例:
火焰围攻 授权: q15040409-creator.功能概述火焰围攻 授权: q15040409-creator.是一项面向实际任务的技能,主要用于Description.;Flask Web. - Python.;核心要点HTTP?;.Version.1.0.tags. wrap.github.automation.。它将相关步骤、工具调用和结果整理方式集中到统一流程中,帮助使用者更快完成目标并减少重复操作。使用与执行使用时应结合输入条件选择合适的执行方式,核对必要参数、依赖环境与输出内容,并按原始要
import httpx
@app.route("/api")
async def api_proxy():
async with httpx.AsyncClient() as client:
resp = await client.get("https://httpbin.org/json")
return resp.json()
注意:不要在全局创建AsyncClient实例并复用——它不是线程安全的,应在每次请求内新建或通过依赖注入管理生命周期。
SQLAlchemy要配async_engine,不能直接用db.session.query()
Flask-SQLAlchemy默认是同步ORM,db.session.query(User).all()会阻塞事件循环。即使你用了Uvicorn,这行代码也会让整个worker停摆。
必须升级到SQLAlchemy 2.0+,并显式配置异步引擎:
- 数据库驱动换成
asyncpg(PostgreSQL)或aiomysql(MySQL) - 创建
create_async_engine,URL前缀为postgresql+asyncpg:// - 用
AsyncSession替代session,所有查询方法加await
片段示意:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("postgresql+asyncpg://...")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
@app.route("/users")
async def list_users():
async with async_session() as session:
users = await session.execute(select(User))
return [u[0].to_dict() for u in users.all()]
before_request这类钩子不支持async,逻辑得收进视图里
你不能写@app.before_request + async def auth_check()——Flask不会await它,只会把协程对象当普通函数调用,导致未消费的协程堆积、内存缓慢增长,甚至引发RuntimeWarning。
可行方案只有两个:
- 把预处理逻辑(如JWT校验、权限检查)直接写在每个
async视图开头,用await调用 - 改用ASGI中间件(如
uvicorn.middleware.proxy_headers.ProxyHeadersMiddleware),它天然支持async
另外注意:g对象在async视图中可用,但它的生命周期仅限当前请求协程,跨await边界没问题;可放心用g.user = await get_current_user()。


















