漫画图解fastapi微服务框架:方老师带你3分钟学会FastAPI中的async/await用法,快速入门异步编程

10年青少年编程教育经验 | 资深Python专家

前大厂技术总监 | 现编程教育创业者

专注青少年编程教育技能提升

@app.route('/items')def get_items():      data = fetch_from_database()    # 等待数据库返回，CPU干等着      return jsonify(data)

@app.get('/items')async def get_items():    data = await fetch_from_database()      # 等待时CPU可以去处理其他请求        return data

from fastapi import FastAPIapp = FastAPI()@app.get('/')async def root():    return {'message':       'Hello World'}@app.post('/users')async def create_user(user:                 UserCreate):    # 异步处理用户创建    return await save_user(user)

import httpxfrom sqlalchemy.ext.asyncio import AsyncSession@app.get('/users/{user_id}')async def get_user(user_id: int,       db: AsyncSession):           # 异步查询数据库       user = await db.get(User, user_id)# 异步调用外部API    async with httpx.AsyncClient() as client:        response = await client.get(           f'https://api.example.com/users/{user_id}')    return {'user': user,           'external_data': response.json()}

import asyncio@app.get('/dashboard')async def get_dashboard():    # 同时发起三个查询，等全部完成    user_count, order_count,         revenue = await asyncio.gather(                get_user_count(),                get_order_count(),                 get_revenue()                )    return {      'users': user_count,      'orders': order_count,       'revenue': revenue   }

@app.get('/wrong')async def wrong():    # 同步阻塞操作会卡住整个事件循环    time.sleep(5)    # 千万别这样！        return {'message': 'done'}

import asyncio@app.get('/correct')async def correct():    await asyncio.sleep(5)      # 使用异步版本的sleep        return {'message': 'done'}

from sqlalchemy.ext.asyncio import       create_async_engine, AsyncSession      # 使用异步数据库驱动（如asyncpg）      engine = create_async_engine(            'postgresql+asyncpg://user:pass@localhost/db')