如何在Python3.10中使用Flask 3.0原生async异步视图函数-Python教程-PHP中文网

flask 3.0 的 async 视图必须搭配 async/await 语法使用，且整个调用链需异步化；混用同步操作会导致事件循环阻塞，须改用 asyncpg、httpx、sqlalchemy 2.0+ 等异步依赖，并通过 asgi 服务器（如 uvicorn）部署。

如何在python3.10中使用flask 3.0原生async异步视图函数

Flask 3.0 的 async 视图函数必须搭配 async/await 语法使用

Flask 3.0 开始原生支持 async 视图函数，但不是“加个 async 就能自动异步”——它要求整个调用链（包括你调用的依赖）也得是异步的。如果你在 async 视图里混用同步阻塞操作（比如 requests.get()、open()、time.sleep()），事件循环会卡住，反而比同步还慢。

常见错误现象：RuntimeWarning: coroutine 'xxx' was never awaited 或请求长时间挂起无响应，本质是协程没被 await 调用，或底层用了同步 I/O。

视图函数必须声明为 async def，返回值仍要是响应对象（str、dict、Response 等），但中间逻辑需 await 异步调用
不能在 async 视图中直接调用同步数据库驱动（如 sqlite3.connect()、psycopg2.connect()），得换 asyncpg、aiosqlite、sqlalchemy 2.0+ with async_engine
模板渲染默认是同步的，render_template() 不支持 async；要用 render_template_string() 配合预编译，或改用 jinja2.AsyncEnvironment

如何正确启动 Flask 3.0 并启用 async 支持

Flask 3.0 默认使用 werkzeug 2.2+，而 Werkzeug 2.2+ 的开发服务器（flask run）已基于 asyncio 重构，但仅当有至少一个 async 视图时才会启用异步模式。没有 async 视图时，它退化为同步行为，不报错也不警告。

关键点：不需要额外配置开关，只要定义了 async 视图，Flask 就自动切换运行时模型；但生产环境必须用支持 ASGI 的服务器（如 uvicorn 或 hypercorn），Werkzeug 自带的 dev server 仅限开发调试。

开发时：直接 flask run 即可，有 async 视图就会走异步路径
生产部署：必须用 uvicorn myapp:app（myapp.py 中定义了 app 实例），不能用 gunicorn + WSGI 模式（它不理解 async 视图）
检查是否生效：启动日志里出现 Running on http://127.0.0.1:5000 (Press CTRL+C to quit) 是同步模式；若看到 Uvicorn running on http://127.0.0.1:8000 才是真正异步

async 视图中调用外部 HTTP API 的正确写法

最典型场景：视图要请求第三方 API。别用 requests（同步阻塞），得换 httpx（原生支持 async）或 aiohttp。否则 await 一个同步函数毫无意义，还会抛出 TypeError: object Response can't be used in 'await' expression。

Python Testing

Python 测试速查：运行 pytest、使用 mock/patch、参数化、fixtures、异步、覆盖率测试。

下载

import httpx
from flask import Flask
<p>app = Flask(<strong>name</strong>)</p><p>@app.route("/api")
async def api_view():
async with httpx.AsyncClient() as client:
resp = await client.get("<a href="https://www.php.cn/link/c2148796071914983ed6b6e9dbbff735">https://www.php.cn/link/c2148796071914983ed6b6e9dbbff735</a>")
return resp.json()
</p>

httpx.AsyncClient() 必须用 async with，不能写成 httpx.AsyncClient() 后直接 await client.get(...)（会漏掉连接池管理）
不要在 async 视图里用 threading 或 concurrent.futures.ThreadPoolExecutor 去“包装”同步函数——这违背异步初衷，且容易引发事件循环竞争
如果必须调用同步函数（如旧版 SDK），可用 await asyncio.to_thread(...)（Python 3.9+），但这是兜底方案，性能不如原生异步

SQLAlchemy 2.0+ 中 async 视图如何查数据库

SQLAlchemy 1.x 完全不支持 async；只有 2.0+ 提供 AsyncEngine 和 AsyncSession。直接沿用老写法（session.query(...).all()）会报 AttributeError: 'AsyncSession' object has no attribute 'query'。

核心变化：查询必须用 select() 构造语句，再通过 session.execute() + scalars() 或 fetchall() 获取结果，且全程 await。

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select
<p>engine = create_async_engine("sqlite+aiosqlite:///app.db")</p><p>@app.route("/users")
async def users():
async with AsyncSession(engine) as session:
stmt = select(User)
result = await session.execute(stmt)
users = result.scalars().all()  # 注意：.all() 是同步方法，但前面 await 已完成执行
return {"users": [u.to_dict() for u in users]}
</p>