Python-FastAPI与数据处理

大类:工程化与云原生 · 共 44 题 · 检索页定位

选择题(22)

q134 · 简单

关于 FastAPI 框架的核心优势,下列描述错误的是?
A. 基于 Python 类型提示自动生成 Swagger/ReDoc 交互式 API 文档
B. 底层基于 Starlette(ASGI)并深度结合 Pydantic,I/O 密集场景性能可与 Node.js、Go 的一线异步框架媲美
C. 原生支持 async/await 异步编程,并内置 OAuth2/JWT、WebSocket 等能力支持
D. FastAPI 内置了完整的 ORM 和数据库迁移能力,因此无需再集成 SQLAlchemy 等第三方库

参考答案要点

• 自动文档:由类型提示 + Pydantic 推断生成 openapi.json,内置 /docs 与 /redoc
• 高性能来源:Starlette 提供 ASGI 异步能力,Pydantic 负责验证与序列化
• 原生 async/await 支持;数据验证、序列化与文档一体化
• FastAPI 不内置 ORM,数据库访问需集成 SQLAlchemy、Tortoise ORM 等;公共逻辑用 Depends 依赖注入复用

来源:[blog.csdn.net](https://blog.csdn.net/HappyAcmen/article/details/146421303)

q135 · 中等

关于 FastAPI 中 async def 与普通 def 路径函数的区别,下列说法正确的是?
A. 普通 def 函数会阻塞事件循环,因此在 FastAPI 中禁止使用
B. async def 函数里执行阻塞的同步调用(如 requests 库、time.sleep)会阻塞整个事件循环,导致所有并发请求一起卡住
C. FastAPI 会自动把 async def 函数中的阻塞同步调用放到线程池里执行
D. 在 I/O 密集的高并发场景下,普通 def 与 async def 的性能表现完全相同

参考答案要点

• async def 由事件循环直接调度,其中的阻塞同步调用会卡住整个事件循环,拖垮所有并发请求
• 普通 def 路径函数会被 FastAPI 自动放入线程池执行,不会阻塞事件循环
• 阻塞库应替换为异步库(httpx.AsyncClient、asyncio.sleep、异步驱动)或用 run_in_executor 包裹
• I/O 密集服务优先 async;CPU 密集任务需要进程池,否则受 GIL 限制占死 worker

来源:[cnblogs.com](https://www.cnblogs.com/xiaomandujia/p/17932445.html)

q136 · 中等

关于 Pydantic 在 FastAPI 中的作用,下列说法错误的是?
A. 请求体声明为 BaseModel 子类后,FastAPI 自动完成解析与类型校验,校验失败返回 422 及字段级错误明细
B. 通过 response_model 可以过滤响应字段(如隐藏密码)并约束响应结构
C. 使用 Pydantic 后,每个路由仍需手写 if-else 逐个判断字段合法性
D. Pydantic 同时承担序列化/反序列化与 OpenAPI schema 生成

参考答案要点

• BaseModel + 类型注解声明式校验,无需手写分支判断;失败自动返回 422 与明细
• response_model 过滤输出字段、固定响应结构,防止敏感字段泄漏
• 同一份模型同时生成 OpenAPI 文档,校验与文档一体化
• Pydantic v2 核心由 Rust 实现,校验性能高;支持嵌套模型与自定义 validator

来源:[blog.csdn.net](https://blog.csdn.net/HappyAcmen/article/details/146421303)

q498 · 中等

FastAPI 路由声明了 response_model=UserOut(不含 password 字段),但函数实际返回的 ORM 对象包含 password 属性。结果是?
A. FastAPI 会按 response_model 对输出做校验与过滤,响应体中不会出现 password,同时 OpenAPI 文档也按 UserOut 展示——这是防止敏感字段外泄的关键机制,返回 dict/ORM 对象时同样生效
B. 运行时报错,必须返回与模型完全一致的 dict
C. password 会照常出现在响应里,只是文档不显示
D. response_model 只用于生成文档,不影响实际响应体

参考答案要点

• 正确答案 A. 官方文档:FastAPI 会 filter 掉输出模型未声明的全部字段,即使函数返回的是包含额外字段的 dict 或数据库对象
• B 支持 dict 与 ORM 对象的自动转换;C/D 把 response_model 误解为纯文档功能
• 配套用法:输入输出用不同模型(UserIn/UserOut),或 response_model_exclude/include 细化

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/response-model/)

q499 · 中等

用 Uvicorn 部署 FastAPI 并希望利用多核 CPU,官方推荐的做法是?
A. 在单个进程里开更多工作线程来并发处理请求
B. 每来一个请求就 spawn 一个子进程处理
C. uvicorn 默认已是多进程,无需任何配置
D. 启动多个 worker 进程(uvicorn --workers N 或 fastapi run --workers N),每个 worker 是独立进程、持有独立事件循环,以此利用多核;但在 Kubernetes 上官方建议每容器只跑单个 Uvicorn 进程,靠 Pod 副本横向扩展

参考答案要点

• 正确答案 D. 官方部署文档:进程复制(worker processes)用于利用多核;K8s 场景推荐容器内单进程+副本扩缩容
• A 单进程多线程受 Python 全局解释器锁限制,无法并行执行 CPU 工作;B 每请求建进程的开销不可接受
• C uvicorn 默认单进程单事件循环

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/deployment/server-workers/)

q500 · 中等

FastAPI 要提供几百 MB 文件的下载接口,下列实现与说法正确的是?
A. 用 open(path).read() 把文件整体读进内存再返回 JSONResponse,最简单也最稳
B. 返回文件必须自己手写 HTTP Range 断点续传逻辑,任何框架都不可能内置支持
C. 应使用 FileResponse(或 StreamingResponse 分块流),由框架/服务器异步流式发送文件,避免把整个文件读入进程内存——大文件整体 read 在并发下会内存暴涨甚至 OOM
D. 下载接口必须写成 async def,并在函数内用阻塞的 open 逐块读

参考答案要点

• 正确答案 C. FileResponse 面向「返回文件」场景,内部以异步方式流式发送,无需把文件实体装进内存
• A 大文件全量进内存是典型反例,并发几个请求就能打爆内存
• B FileResponse 已处理 etag/last-modified 等响应语义(Range 在新版 Starlette 亦有支持);D 在 async def 里做阻塞读会卡住事件循环,恰恰应用 FileResponse 避免

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/advanced/custom-response/)

q501 · 中等

FastAPI 的 async 路由里,每个请求都 httpx.AsyncClient() 新建客户端去调第三方接口。压测后出现大量 TIME_WAIT 且 RT 居高。正确修复是?
A. 在应用 lifespan 中创建全局 AsyncClient(内建连接池,复用 TCP/TLS 连接),请求全程复用这一实例
B. 把 async 路由改成普通 def,让线程池去每次新建客户端
C. 把客户端 timeout 设为 0,让请求永不超时
D. 这是 Python 的固有缺陷,无法优化

参考答案要点

• 正确答案 A. Client 实例的连接池会 keep-alive 复用连接;每次新建客户端等于每次重新 TCP+TLS 握手,高频下延迟放大、本地端口耗尽(TIME_WAIT 堆积)
• B 换线程池不消除建连开销,反而引入线程开销
• C timeout=0 表示禁用超时,慢请求会无限堆积;D 属于典型反模式且有标准解法

来源:[python-httpx.org](https://www.python-httpx.org/advanced/clients/)

q502 · 中等

pandas 给 2000 万行 DataFrame 新增一列「单价×数量」,下列实现与快慢判断正确的是?
A. df.apply(axis=1) 逐行最快,是官方推荐写法
B. 手写 for 循环遍历 DataFrame 最快,解释器开销最小
C. 向量化运算 df['amount'] = df['price'] * df['qty'] 最快(整列在 C 实现层完成);iterrows 逐行最慢(每行都要构造 Series 对象);apply 每行调用一次 Python 函数,远慢于向量化——能向量化就不要逐行处理
D. 三者性能相同,只是代码风格差异

参考答案要点

• 正确答案 C. 官方性能指南:向量化 > itertuples/numpy > apply > iterrows,量级差距可达数十到上千倍
• A/B 逐行 Python 层解释执行与对象构造的开销远高于 C 层整列运算
• D 性能差异是 pandas 优化的第一课

来源:[pandas.pydata.org](https://pandas.pydata.org/docs/user_guide/enhancingperf.html)

q503 · 困难

FastAPI 路由直接返回 SQLAlchemy ORM 对象,期望 Pydantic 按声明模型序列化输出。Pydantic v2 下需要的配置是?
A. 必须放弃 response_model,改用 dataclass 返回
B. 在模型的 Config 中声明 from_attributes=True(v1 时代叫 orm_mode=True),Pydantic 才会从对象属性(attribute)读取并校验;默认模式下 Pydantic 按 dict 输入校验,直接喂 ORM 对象会校验失败
C. 无需任何配置,Pydantic 天然能从任意对象取属性
D. 把 SQLAlchemy 模型改为继承 BaseModel 即可

参考答案要点

• 正确答案 B. from_attributes 允许「从任意对象的属性构造模型」,是 FastAPI 官方 SQL 教程中返回 ORM 对象的标准配置
• C v2 默认输入按 dict/关键字参数校验,ORM 对象会报 validation error
• D 声明方向反了:序列化模型是 Pydantic 模型,不是让 ORM 模型继承;A 与 response_model 可正常配合

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/sql-databases/)

q504 · 简单

FastAPI 自带的交互式文档 /docs 与 /openapi.json 是怎么来的?
A. 需要开发者手写 Swagger YAML 或注解
B. 文档在首次启动时联网抓取模板生成,离线环境不可用
C. 只能在开发环境开启,生产环境必须关闭
D. 由路由装饰器中的类型注解、Pydantic 模型与参数声明自动生成 OpenAPI Schema,/docs 只是基于该 schema 渲染的 Swagger UI(另有 /redoc);改代码即同步更新,无需单独维护文档

参考答案要点

• 正确答案 D. FastAPI 基于 Python 类型注解自动产出 OpenAPI schema,文档与代码天然一致
• A 正是传统框架的痛点,FastAPI 的卖点是零手写;B/C 无依据
• 生产环境是否暴露文档路由可自行控制,但这属于部署选择而非框架限制

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/first-steps/)

q505 · 简单

FastAPI 路径函数参数的来源推断规则,正确的是?
A. 任何函数参数都会被解析为请求体字段
B. 标量类型参数(int/str/bool 等)默认是请求头字段
C. 参数名出现在路径占位符(如 /items/{item_id})中的是路径参数;其余标量类型参数默认按查询参数(query)解析;参数类型是 Pydantic 模型时才默认解析为请求体(body)
D. 参数来源没有任何自动推断,必须逐个用装饰器显式声明

参考答案要点

• 正确答案 C. FastAPI 按路径占位符匹配与类型注解决定参数来源,如 ?skip=0&limit=10 对应两个 query 参数
• A 请求体只留给 Pydantic 模型等复杂类型;B 标量默认是 query 不是 header
• D 自动推断是核心特性;需要覆盖默认时才用 Query()/Path()/Body() 显式声明

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/query-params/)

q645 · 中等

关于 FastAPI(Starlette)中间件,下列说法错误的是?
A. 用 @app.middleware("http") 写法基于 BaseHTTPMiddleware 实现,可以在请求进入路由前后插入自定义逻辑
B. 自定义纯 ASGI 中间件通常比 BaseHTTPMiddleware 开销更小,对 StreamingResponse 等流式响应的兼容性也更好
C. 中间件与 Depends 依赖都能做鉴权,区别之一是中间件工作在 ASGI 层拿不到路径操作的参数类型信息,而依赖可以与 OpenAPI 文档、422 错误体系更好地集成
D. 多个中间件按添加顺序执行:先添加的中间件最先处理请求,响应时也最先处理返回

参考答案要点

• D 错误:中间件是洋葱模型,后添加的位于外层、最先处理请求;响应沿相反方向返回,先添加的中间件最先 outward 处理响应
• A 正确:BaseHTTPMiddleware 提供 call_next 抽象;注意部分旧版本 BaseHTTPMiddleware 与流式响应、后台任务存在已知兼容问题
• B 正确:纯 ASGI 中间件直接操作 scope/receive/send,没有中间抽象层,开销小且行为可控
• C 正确:选型经验——全局横切关注点(日志、CORS、trace id)用中间件;与业务参数、文档相关的鉴权(如按路由鉴权、分页参数)用依赖注入
• 常用内置中间件:CORSMiddleware、GZipMiddleware、TrustedHostMiddleware
• 排障:中间件内抛出的异常不会走 HTTPException 全局处理器,需要自己捕获并组织错误响应

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/middleware/)

q648 · 中等

关于 Pydantic v2 在 FastAPI 中的校验行为,下列说法错误的是?
A. 用 Field(gt=0, le=100) 声明数值范围,校验失败返回 422,响应体含逐字段的错误详情
B. 请求体 JSON 语法正确但字段校验失败时返回 400 Bad Request;只有 JSON 格式错误才返回 422
C. 用 @field_validator 可自定义字段校验逻辑,校验器中抛出 ValueError 会转为 422 响应
D. model_config = ConfigDict(extra="forbid") 可以让请求体携带未定义字段时直接校验失败

参考答案要点

• B 错误:请求体 JSON 解析失败和字段校验失败都返回 422 Unprocessable Entity;400 通常留给业务层自定义错误码
• A 正确:数值/字符串长度等约束由 Field 声明,错误详情在 detail 数组中逐项给出 loc/msg/type
• C 正确:@field_validator 支持 before/after/wrap 模式;v1 的 @validator 已被替代
• D 正确:extra 可配置 ignore(默认,忽略多余字段)/forbid(报错)/allow(保留并透传)
• 工程实践:响应侧用 response_model 过滤输出字段防止内部字段泄露;alias 处理驼峰与下划线映射
• v2 性能:校验核心由 Rust 实现(pydantic-core),比 v1 快一个量级,大模型结构化输出解析场景受益明显

来源:[docs.pydantic.dev](https://docs.pydantic.dev/latest/concepts/validators/)

q654 · 简单

关于 FastAPI 的 BackgroundTasks 与专业任务队列(如 Celery/ARQ/Dramatiq)的选型,下列说法错误的是?
A. BackgroundTasks 的任务在响应返回后于当前进程内执行,进程重启时未完成的任务会丢失
B. 发送通知、写审计日志、推送埋点等轻量收尾动作适合用 BackgroundTasks,避免引入额外中间件
C. 任务需要重试、延迟执行、定时调度、分布式扩展或可靠投递时,应选用 Celery 等任务队列
D. Celery worker 必须与 FastAPI 应用部署在同一个容器中,通过共享内存通信

参考答案要点

• D 错误:Celery worker 是独立进程、独立部署,通过消息中间件(Redis/RabbitMQ)解耦通信,这正是它能独立扩展与容灾的前提
• A 正确:进程内执行意味着无持久化,滚动发布或重启会丢任务,可靠场景必须外置队列
• B 正确:轻任务进程内完成,减少运维复杂度;注意别放重 IO/CPU 任务,且异常默认只打日志,要自己加异常上报
• C 正确:at-least-once 投递语义、重试退避、定时任务、专属 worker 池是任务队列的核心价值
• 补充:async 生态里 ARQ/RQ 与 asyncio 适配更好;大模型场景常见「API 收任务入队,worker 拉起推理」的异步任务模式
• 排障视角:BackgroundTasks 抛错不影响已返回的响应,线上要以日志/监控确认任务真的执行了

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/background-tasks/)

q828 · 困难

关于 SQLAlchemy 2.0 的写法与理念,下列说法错误的是?

A. 旧版 session.query(User).filter(...) 写法仍可用,但 2.0 风格主推 select(User).where(...) 配合 session.execute()/session.scalars()
B. 声明模型改为继承 DeclarativeBase 的基类,取代旧版 declarative_base() 函数的做法
C. Session 采用 unit of work 模式:add/修改对象只是累积变更,直到 flush/commit 才真正向数据库发出 SQL
D. Session.commit() 之后对象属性默认仍保持可用且不过期,后续读取属性不会再触发任何数据库查询

参考答案要点

• 正确答案 D. 默认 expire_on_commit=True,commit 后对象属性全部过期,下次访问会触发刷新(重新 SELECT),Session 已关则抛 DetachedInstanceError;Web 请求内常设 expire_on_commit=False 规避
• A/B/C 都是准确的 2.0 迁移要点:统一 select() 风格、DeclarativeBase 基类、unit of work 延迟落库
• D 的坑在 FastAPI 中常见:commit 后再序列化 ORM 对象时突然多发查询或直接报错,根因就是属性过期刷新机制

来源:[docs.sqlalchemy.org](https://docs.sqlalchemy.org/en/20/tutorial/orm_data_manipulation.html)

q829 · 中等

关于 SQLModel,下列说法正确的是?

A. SQLModel 是 FastAPI 作者在 Pydantic 与 SQLAlchemy 之上做的一层封装,table=True 的模型类既是 SQLAlchemy 表模型又是 Pydantic 校验模型
B. SQLModel 完全取代 SQLAlchemy,自带独立的 SQL 引擎实现,不再依赖 SQLAlchemy
C. 写不写 table=True 行为完全一样,两种类都会对应数据库表
D. 使用 SQLModel 后就不能再定义单独的响应模型,必须把 ORM 模型原样暴露给 API

参考答案要点

• 正确答案 A. 一个类同时是 SQLAlchemy 模型与 Pydantic 模型,可同时用作请求/响应 schema 与表模型
• B 错:底层就是 SQLAlchemy 的 engine/Session,SQLModel 只是薄层
• C 错:不带 table=True 的类是纯 Pydantic 数据模型,不会建表
• D 错:可以再定义不含敏感字段的读取模型(继承同一基类)做输出过滤

来源:[sqlmodel.tiangolo.com](https://sqlmodel.tiangolo.com/)

q830 · 困难

SQLAlchemy 关系查询中的 N+1 问题与预加载策略,下列说法错误的是?

A. 默认 lazy loading 下,查出 100 个订单后循环访问 .user 属性,会产生 1 条查订单 + 100 条查用户的 SQL
B. selectinload 通过一条带 IN 的附加查询批量取回全部关联对象,常用于一对多/集合关系,避免 JOIN 行数放大
C. joinedload 把关联表 JOIN 进主查询一次取回,适合多对一;一次 JOIN 多个集合关系则可能造成行数笛卡尔积膨胀
D. 只要模型声明了 relationship,SQLAlchemy 就会自动分析查询并为开发者选择最优加载策略,无需显式指定

参考答案要点

• 正确答案 D. relationship 默认就是 lazy loading,不会自动切换 eager 策略,N+1 需要开发者显式用 selectinload/joinedload(或 lazy='selectin')解决
• A 准确描述了 N+1 的成因;B/C 是两种预加载的典型取舍:selectinload 防行数放大、joinedload 减少 SQL 次数
• 面试加分点:批量接口(列表页)是最容易踩 N+1 的地方,压测或开 SQL echo 能直接看到 SQL 条数暴涨

来源:[docs.sqlalchemy.org](https://docs.sqlalchemy.org/en/20/tutorial/)

q831 · 困难

FastAPI + SQLAlchemy 的生产部署中,关于 Session 与连接池,下列说法错误的是?

A. 每个请求应使用独立 Session(常用 yield 依赖注入创建/关闭),而不是全局共用一个 Session
B. create_engine 的 pool_size/max_overflow 分别控制连接池常驻大小与可溢出的临时连接数;engine 应是进程级单例,随应用生命周期创建
C. pool_pre_ping=True 会在从池里取出连接前做一次轻量检测,避免拿到已被数据库或中间件断开的失效连接
D. 连接池越大越好,把 pool_size 设成远超数据库可承载连接数可以显著提升吞吐

参考答案要点

• 正确答案 C. pool_pre_ping 以每次取连接多一次轻量检查的代价,换掉「8 小时闲置后第一请求报连接已断」这类经典故障
• D 错:连接池过大会打满数据库连接数与内存,合理上限要看数据库 max_connections 与「worker 进程数 × pool_size」的总账
• A/B 是标准实践:请求级 Session + 进程级单例 engine,K8s 里总连接数 = 副本数 × worker 数 × (pool_size + max_overflow)

来源:[docs.sqlalchemy.org](https://docs.sqlalchemy.org/en/20/core/pooling.html)

q832 · 中等

关于 pytest 的 fixture 机制,下列说法正确的是?

A. 测试函数通过同名参数声明使用 fixture,pytest 自动查找注入;conftest.py 里的 fixture 对同目录及子目录测试自动可见,无需 import
B. fixture 的 scope 只有 function 一种,每个测试执行前都会重新跑一遍 fixture
C. scope="session" 的 fixture 在每个测试函数执行前后都会重新建立一次
D. yield 式 fixture 的清理(teardown)代码写在 yield 之前、fixture 主体代码之后执行

参考答案要点

• 正确答案 A. 参数名注入 + conftest.py 自动发现是 pytest fixture 的两大基本设计
• B 错:scope 有 function/class/module/package/session 五级,逐级复用
• C 错:session 级整个测试会话只建立一次;D 错:teardown 在 yield 之后执行,yield 前是 setup

来源:[docs.pytest.org](https://docs.pytest.org/en/7.4.x/explanation/fixtures.html)

q833 · 困难

关于 asyncio.gather 与 TaskGroup(3.11+),下列说法错误的是?

A. gather 默认情况下,某个任务抛异常会立刻向 await gather 的调用方传播,但其余已启动的任务不会被自动取消、仍继续运行
B. gather(return_exceptions=True) 会把子任务异常当作结果放进返回列表,由调用方统一处理,适合「尽力并发」场景
C. TaskGroup 中任一任务失败会取消同组其余任务,并把异常合并为 ExceptionGroup 抛出,是结构化并发的体现
D. gather 的返回结果列表按任务完成的先后时间排序,而不是按传入参数的顺序

参考答案要点

• 正确答案 D. gather 返回列表严格按传入 awaitable 的顺序排列,与完成先后无关——这也是它适合「并发请求再按序装配结果」的原因
• A 是高频误点:默认模式下异常传播但其余任务继续跑,想全部取消要么自己收尾要么用 TaskGroup
• B/C 准确:return_exceptions 换来「不互相牵连」,TaskGroup 换来「一损俱损」,按业务语义选择

来源:[docs.python.org](https://docs.python.org/3/library/asyncio-task.html)

q834 · 中等

关于 CPython 的 GIL(全局解释器锁),下列说法正确的是?

A. 有了 GIL,多线程做纯 CPU 密集计算也能随核数线性加速,完全不需要 multiprocessing
B. GIL 保证同一时刻只有一个线程执行 Python 字节码;线程进入阻塞 I/O(如 socket 读、time.sleep)时会释放 GIL,因此多线程仍能加速 I/O 密集任务
C. numpy/pandas 等库的重计算段从不释放 GIL,所以任何情况下都无法并行
D. GIL 是操作系统层面强制的锁,与 CPython 实现无关,PyPy 等其他实现的行为完全一致

参考答案要点

• 正确答案 B. GIL 是 CPython 实现细节:同刻仅一个线程执行字节码,线程做阻塞 I/O 与许多 C 扩展重计算时会释放 GIL,所以多线程对 I/O 密集仍有效
• A 错:CPU 密集多线程无法并行,甚至因切换更慢,要 multiprocessing/多进程
• C 错:numpy/pandas 大量 C 实现的重计算段会释放 GIL,可并行;D 错:GIL 是 CPython 的实现细节而非语言规范,不是所有实现都有

来源:[docs.python.org](https://docs.python.org/3/glossary.html)

q835 · 中等

pandas 的 groupby 操作中,下列说法错误的是?

A. agg('mean') 返回以分组键为索引、每组一个值的聚合结果
B. transform('mean') 返回与原 DataFrame 等长、索引与原行对齐的结果,适合直接生成「组均值」新列
C. apply 逐组执行自定义函数,返回形状随函数而定,可以是标量、Series 或 DataFrame
D. filter(lambda g: ...) 是在每组内部按条件筛选出部分行,不会整组保留或整组丢弃

参考答案要点

• 正确答案 D. groupby.filter 以组为单位决策:条件为 True 的组整组保留、False 的组整组丢弃,不是组内筛选行(组内筛行应先 filter 再用布尔索引/transform 配合)
• A/B/C 是三个常用出口的准确区分:agg 收敛、transform 保持等长对齐、apply 自由形状
• transform 是「组统计广播回原行」的惯用法,比 apply+merge 更快也更不易错

来源:[pandas.pydata.org](https://pandas.pydata.org/docs/user_guide/groupby.html)

简答题(17)

q137 · 中等

FastAPI 为什么性能高?请从架构层面说明性能来源,并指出它在哪些场景下并不快。

参考答案要点

• 基于 Starlette 的 ASGI 异步架构,高并发 I/O 下事件循环避免线程创建与切换开销
• Pydantic 基于类型注解做验证与序列化,v2 核心用 Rust 实现,开销低
• 原生 async/await + uvicorn 等异步服务器组合
• 不快的场景:async def 里写阻塞调用会卡死事件循环;CPU 密集任务受 Python GIL 限制,须用进程池
• 与 Flask 等 WSGI 同步框架相比的优势主要体现在 I/O 密集场景,而非计算密集场景

来源:[cnblogs.com](https://www.cnblogs.com/xiaomandujia/p/17932445.html)

q138 · 中等

FastAPI 的依赖注入系统(Depends)是什么?适合用来做哪些事情?

参考答案要点

• 机制:定义依赖(函数或类)→ 路由参数用 Depends() 声明 → 请求到达时自动解析并注入
• 典型用途:数据库会话管理、JWT/权限校验、公共参数解析(分页等)的复用
• 支持依赖嵌套(依赖链)与异步依赖
• yield 依赖可在请求结束时执行释放逻辑(finally),适合管理会话/连接生命周期
• 降低耦合、便于单元测试(测试时替换依赖)

来源:[cnblogs.com](https://www.cnblogs.com/xiaomandujia/p/17932445.html)

q139 · 中等

FastAPI 与 Flask、Django 有什么区别?什么样的项目适合选 FastAPI?

参考答案要点

• FastAPI:ASGI 原生异步、类型驱动、自动文档,定位高性能 API 服务与微服务
• Flask:WSGI 同步微框架,灵活轻量,生态成熟,异步能力需借助扩展
• Django:batteries-included 全家桶(ORM/Admin/模板/表单),适合快速建站与管理后台
• 选型维度:I/O 密集高并发 API、机器学习模型服务 → FastAPI;重模板渲染的传统 Web/后台 → Django;小型工具 → Flask
• 现实约束:团队熟悉度、存量生态与组件积累也是重要因素

来源:[cnblogs.com](https://www.cnblogs.com/xiaomandujia/p/17932445.html)

q140 · 中等

FastAPI 中如何做统一规范的异常处理与错误响应?

参考答案要点

• 用 HTTPException 抛出标准错误,携带状态码与 detail
• 通过 @app.exception_handler() 注册全局异常处理器,统一错误响应结构
• 自定义业务异常类并绑定对应 handler,路由内直接 raise
• 可用 include_router 为路由组绑定异常处理;校验错误默认 422,可通过 handler 改写为统一格式
• 避免裸 try/except 吞掉异常;区分业务异常与系统异常的返回语义

来源:[cnblogs.com](https://www.cnblogs.com/xiaomandujia/p/17932445.html)

q141 · 中等

FastAPI 如何实现文件上传和轻量后台任务?大文件上传需要注意什么?

参考答案要点

• 文件上传:参数声明为 UploadFile(配合 File),用 await file.read() 读取,大文件需分块读写
• 上传校验:文件类型、大小限制,用完调用 close() 释放资源
• 后台任务:BackgroundTasks 在响应返回后执行,适合发通知、写日志等轻量操作
• 重任务(转码、批处理)不适合 BackgroundTasks,应交给独立任务队列(worker)
• 分片/断点续传与幂等(已传分片记录)、短期上传凭证、孤立分片清理

来源:[blog.csdn.net](https://blog.csdn.net/HappyAcmen/article/details/146421303)

q646 · 中等

FastAPI 应用启动时需要把 2GB 的机器学习模型加载进内存、初始化数据库连接池,关闭时需要释放这些资源。请说明如何用 lifespan 机制实现,以及为什么官方不再推荐 @app.on_event("startup"/"shutdown")。

参考答案要点

• 写法:用 asynccontextmanager 装饰 lifespan 函数,yield 之前是启动逻辑、之后是清理逻辑,通过 lifespan 参数传入 FastAPI 实例
• 资源共享:启动时创建的对象挂到 app.state(如 app.state.model),路由里通过 request.app.state 取用
• 执行时机:startup 逻辑在开始接收请求之前执行完毕,清理逻辑在处理完存量请求之后执行;多个资源的进入与退出按上下文管理器嵌套顺序执行
• on_event 弃用原因:多个 startup/shutdown 处理器分散注册、无法在生命周期内共享局部状态、与 ASGI lifespan 协议的对应关系不直观;lifespan 把整个生命周期收敛为一个上下文
• 工程注意:模型加载放启动阶段而非模块导入顶层,保证多 worker 各自加载且失败在启动时暴露;惰性加载需自行加锁防止并发重复加载
• 关联:uvicorn/gunicorn 的优雅停机会先等存量请求结束再触发 lifespan 清理,清理逻辑要幂等且快

来源:synthesized

q647 · 中等

FastAPI 依赖注入系统中的「yield 依赖」是什么?它解决什么问题?同一个请求内多个位置 Depends 同一个依赖函数时默认行为是什么,如何关闭?

参考答案要点

• yield 依赖:依赖函数写成生成器,yield 出的值交给路由使用,yield 之后的代码在响应发送后执行,适合数据库会话这类需要请求级清理的资源
• 清理保证:即使路由抛异常,yield 之后的代码也会执行(类似 finally);需要抛给客户端的 HTTPException 应放在 yield 之前
• 依赖缓存:默认同一请求内同一依赖只执行一次(use_cache=True),多处 Depends 复用同一结果;传 use_cache=False 关闭
• 典型用法:DB 会话(请求结束自动关闭/回滚)、事务边界管理、认证上下文对象
• 嵌套依赖:依赖可以再 Depends 其他依赖形成树状解析,子依赖的清理在外层依赖清理之前完成
• 与 lifespan 分工:全局单例(连接池、模型)放 lifespan,请求级资源放 yield 依赖;不要在依赖里反复创建重资源

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/tutorial/dependencies/)

q649 · 困难

你有 Go 开发背景,现在转 Python 写 FastAPI 服务。请对比 Python asyncio 与 Go 并发模型的差异,并说明这对写异步 Python 服务意味着什么(GIL 在其中扮演什么角色)。

参考答案要点

• 调度模型:asyncio 是单线程协作式调度,协程只在 await 点让出控制权,一个协程不yield就独占事件循环;goroutine 是运行时抢占式调度(M:N 映射到 OS 线程),阻塞一个 goroutine 不影响其他
• GIL 的影响:CPython 全局解释器锁使同一进程任一时刻只有一个线程执行 Python 字节码,多线程无法并行 CPU 任务;asyncio 规避不了 GIL——它优化的是 IO 并发而不是 CPU 并行
• 阻塞代价差异:Python 协程里调用同步阻塞库(CPU 重计算、同步 IO)会卡住整个事件循环上的所有请求;Go 里同步调用只占用当前 goroutine 所在线程,由 runtime 调度其他任务
• 成本与生态:goroutine 初始栈 KB 级、开十万级无压力;Python 协程也轻量,但存在「async 感染」问题——调用链上只要有一个同步库就要用线程包装,心智负担大
• 工程含义:IO 密集用 asyncio 加全链路 async 库(httpx/asyncpg),CPU 密集丢进程池或写原生扩展;Go 的 channel 对应 asyncio.Queue,errgroup 对应 TaskGroup
• 迁移提示:别把 Go 直觉带进来——同步调用便宜这一前提在 asyncio 里不成立,代码 review 重点看 await 链路里的隐藏阻塞点

来源:synthesized

q651 · 中等

线上 FastAPI 服务偶发所有请求一起卡顿数秒。排查发现某 async def 路由里调用了同步的 requests 库,另一个路由里有正则处理超大文本。解释为什么会「一颗老鼠屎坏一锅粥」,并给出至少 3 种修复手段。

参考答案要点

• 根因:async def 路由运行在事件循环线程上,同步 requests 阻塞的是整个事件循环——所有协程都无法调度;CPU 密集的正则回溯同理。而普通 def 路由会被 FastAPI 自动丢进线程池执行,反而不会卡事件循环
• 修复一:换 async 客户端(httpx.AsyncClient/aiohttp),彻底异步化 IO
• 修复二:必须同步的库用 await asyncio.to_thread(fn) 或 starlette 的 run_in_threadpool 丢线程池
• 修复三:CPU 密集任务丢进程池(ProcessPoolExecutor),或拆成独立 worker 服务解耦
• 辅助手段:接口与下游调用都设超时;正则预编译、限制输入长度、避免灾难性回溯写法;用 asyncio 的慢回调日志(slow callback duration)定位阻塞点
• 预防:CI 里扫描 async 函数对 requests、time.sleep 等已知阻塞库的引用

来源:synthesized

q652 · 中等

你要用 FastAPI 给前端提供大模型流式输出接口(SSE)。说明实现要点和至少 4 个生产环境注意事项。

参考答案要点

• 实现:StreamingResponse(media_type 为 text/event-stream)接收 async generator,逐 token yield;或用 sse-starlette 的 EventSourceResponse 处理 SSE 帧格式细节
• 生成器内 async for 消费模型 SDK 的流式接口;客户端断开时生成器被关闭,要捕获并取消上游请求,避免白白烧 token
• 注意一:Nginx/网关必须关缓冲(proxy_buffering off、响应头 X-Accel-Buffering 为 no),否则前端表现为卡住后一次性吐出
• 注意二:长连接超时——网关 read timeout、uvicorn 超时与重启策略;用注释行(: ping)做心跳保活
• 注意三:并发与背压——慢客户端会积压发送队列,要限制单实例并发流数,超出的排队或降级
• 注意四:鉴权与计费在握手阶段完成;流中发生错误用 SSE 的 event:error 帧通知而不是改状态码;流结束后异步记录 token 用量

来源:synthesized

q655 · 中等

你的服务调用大模型 API 要求返回 JSON,但线上经常出现:返回被 markdown 代码块包裹(json 围栏)、字段名大小写不一致、必填字段缺失、字符串里带未转义换行导致 json.loads 失败。请设计一个基于 Pydantic 的鲁棒解析层。

参考答案要点

• 预清洗:剥离 markdown 围栏、截取首个左花括号到末个右花括号的片段、去 BOM 与首尾空白;严格解析失败再降级(修复库或带报错二次请求模型)
• 结构化校验:定义目标 BaseModel,json.loads 后用 model_validate;ValidationError 精确给出出错字段的 loc;用 alias 配置兼容字段名大小写与别名
• 容错分级:可修复问题(别名映射、类型强转)在 validator 的 before 模式静默修复;不可修复的触发重试——把报错信息回传给模型要求修正,设最大重试次数与退避
• 输出侧配合:调用侧优先用 response_format/json 模式或函数调用强制结构化,降低解析层压力;温度调低减少发挥
• 观测:统计解析失败率、静默修复介入率、重试带来的 token 成本,作为 prompt 质量指标
• 兜底:重试仍失败返回明确错误码给上游,禁止把半解析对象往下游传

来源:synthesized

q836 · 困难

生产部署 FastAPI 常用 gunicorn -k uvicorn.workers.UvicornWorker --workers N 或 uvicorn --workers N。请说明这种多进程 worker 模型:master 与 worker 各自的职责是什么?单个 worker 崩溃后会发生什么?--preload 预加载有什么利弊?worker 数量怎么取?

参考答案要点

• master 进程不处理请求:负责绑定端口、pre-fork 出 N 个 worker 子进程、监控存活;worker 异常退出后 master 自动拉起新 worker 补足数量
• 每个 worker 是独立进程,各持有事件循环与内存,单 worker 卡死/崩溃不影响其他 worker,进程级隔离绕开 GIL 利用多核
• --preload 在 fork 前加载应用:省内存(子进程写时复制共享)、加快启动;但与 --reload 开发模式冲突,且 fork 后要小心连接池/锁等带状态对象
• worker 数量通常从 CPU 核数起步做参考,IO 密集可更高,最终以压测校准;K8s 中「副本数 × worker 数」才是总并发,worker 过多会与 Pod CPU limit 相互挤压

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/deployment/server-workers/)

q837 · 中等

asyncio 中 wait_for / asyncio.timeout(3.11+)超时后,内部任务会发生什么?若某段清理逻辑不希望被外部取消波及,用什么原语?gather(return_exceptions=True) 与 TaskGroup 在「部分失败」处理上的取向有何不同?

参考答案要点

• wait_for(aw, t) 超时会先向内部任务发出取消、等它真正结束,再向调用方抛 TimeoutError;asyncio.timeout 上下文同理,TimeoutError 要在上下文外捕获
• 不希望被取消的清理用 asyncio.shield(aw) 保护:等待方收到 CancelledError,被保护任务继续执行完毕;普通清理放 finally 且要快
• gather(return_exceptions=True):异常作为结果返回、不取消其余任务,适合「尽力而为聚合」;TaskGroup:一失败即取消全组、抛 ExceptionGroup,适合「失败要整体失败」的结构化并发
• 补充:wait 自身超时不取消任务、返回 done/pending 集合,适合竞速类场景

来源:[docs.python.org](https://docs.python.org/3/library/asyncio-task.html)

q838 · 中等

Python 里 CPU 密集任务并行为什么常用 multiprocessing/ProcessPoolExecutor 而不是 threading?多进程有哪些额外代价?什么时候应该改用 Celery 这类分布式任务队列?

参考答案要点

• GIL 使多线程无法并行执行字节码,CPU 密集场景多线程甚至比单线程慢(切换开销);多进程各自持有解释器和 GIL,能真正用满多核
• 代价:进程启动贵;任务参数与结果必须可 pickle,跨进程序列化有成本;内存不共享,靠队列/共享内存 IPC;池大小要匹配核数与内存
• 任务分钟级以上、需要跨机器水平扩展、削峰填谷、失败重试与结果持久化时,进程池不够用,选 Celery/ARQ/Dramatiq 等分布式队列
• 进程池定位是单机、进程内的秒级并行;Web 服务内联跑重任务还会拖垮请求,应外置到 worker

来源:synthesized

q839 · 中等

pd.read_csv(path, chunksize=100_000) 与直接 read_csv 的本质区别是什么?用它做「全表去重 + 按城市统计」时要注意什么?还能配合哪些手段进一步降低内存?

参考答案要点

• 返回 TextFileReader 迭代器逐块惰性读取,每次只有一块 DataFrame 进内存,峰值内存约等于单块;iterator=True 等价,还可用 get_chunk
• 每个 chunk 相互独立,跨块状态要自己维护:全局去重要么循环外维护 seen 集合、要么各块先局部去重再 concat 后整体 drop_duplicates;分组统计可用增量累加或最后统一 groupby
• 配合 dtype 显式指定列类型(避免 object 推断)、usecols 只读需要的列、大基数字符串列转 category
• 注意:chunksize 仅 C 引擎支持,engine='pyarrow' 时传它会报错;合并各块结果用 pd.concat

来源:[pandas.pydata.org](https://pandas.pydata.org/docs/user_guide/io.html)

q840 · 简单

说明 venv + pip freeze 的传统依赖管理做法与 poetry/uv 这类工具的差别。为什么要有 lock 文件?requirements.txt 不固定版本会有什么问题?

参考答案要点

• venv 创建隔离环境,pip freeze > requirements.txt 固化当前环境的全部已装版本:可复现但靠人工维护,快照含全部间接依赖,无跨平台解析结果保证
• poetry/uv 基于 pyproject.toml 声明直接依赖,解析后生成 lock 文件锁定完整依赖树(含哈希),任何机器安装结果一致;uv 兼容 pip 用法且解析安装更快
• 不固定版本:构建不可复现,今天和上周装的依赖不同,「在我机器上是好的」类事故的常见来源
• 惯例:应用程序要 lock(可复现优先),开源库只声明版本范围(兼容性优先),两者不要混

来源:[docs.astral.sh](https://docs.astral.sh/uv/)

q841 · 中等

python -m build 会产出什么?wheel 与 sdist 有什么区别?企业内部如何搭建私有 Python 包分发?

参考答案要点

• 产出 sdist(.tar.gz 源码分发)和 wheel(.whl 预构建二进制分发)到 dist/ 目录
• wheel 免构建步骤安装快,pip 优先选 wheel,含 C 扩展的平台 wheel 免去用户本机编译;wheel 失败或需要从源码构建时回退 sdist
• pyproject.toml 是现代标准:[project] 元数据 + [build-system] 构建后端;发布用 twine upload 到 PyPI/私有索引
• 私有分发:devpi/Nexus/Artifactory 建私有索引(可代理缓存官方源),pip install -i 指定 index-url,内部公共库走私有仓版本化复用

来源:[packaging.python.org](https://packaging.python.org/en/latest/tutorials/packaging-projects/)

场景题(5)

q142 · 困难

你用 FastAPI 提供一个"导出全部订单"的接口,数据量 500 万行。当前实现同步查询并一次性返回,30 秒超时且把数据库和其他请求拖垮。请重新设计这个导出能力。

参考答案要点

• 改异步任务:创建导出任务立即返回任务 ID,后台 worker 处理,前端轮询或通知获取结果
• 读数据用游标分页(keyset,WHERE id > last_id)避免 LIMIT/OFFSET 深分页拖垮数据库
• 流式处理:分批读取、分批写入文件,控制内存占用;必要时用 StreamingResponse 做小规模流式下载
• 结果上传对象存储,返回带有效期和权限控制的下载链接
• 限制并发导出任务数、任务创建幂等,防止滥用
• 明确快照语义:接受导出期间数据变化,或用数据库快照保证一致;trade-off:同步实时体验 vs 异步复杂度、新鲜度 vs 性能、主库压力 vs 从库成本

来源:[interview.javaguide.cn](https://interview.javaguide.cn/system-design/system-design-and-scenario-questions.html)

q143 · 困难

你用 FastAPI 封装了一个机器学习预测服务,模型单次推理约 800ms 且为纯 CPU 计算。压测发现并发一升高,所有请求的延迟一起暴涨到十几秒。请分析原因并给出改造方案。

参考答案要点

• 根因分析:推理是 CPU 密集的同步调用,若写在 async def 中会阻塞事件循环,所有请求被串行排队;GIL 使单进程无法并行计算
• 改造一:路由保持 async,把推理放入 ProcessPoolExecutor(run_in_executor)绕开 GIL
• 改造二:多 worker 部署(uvicorn/gunicorn,worker 数对齐 CPU 核数)水平扩展
• 保护措施:并发信号量/限流、请求超时,防止过载雪崩;可引入排队与批处理(batch)提升整体吞吐
• 模型权重只加载一次全局复用,避免每请求重复加载
• trade-off:进程池的进程开销与内存(每个进程一份模型) vs 延迟收益;批处理提高吞吐但增加单请求等待

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/zh/)

q650 · 困难

FDE 交付场景:你的 FastAPI 接口要聚合客户内网 5 个系统(工单/库存/客户画像/工时/知识库)的数据,各接口耗时 300-800ms 不等,当前串行调用合计约 3 秒,客户要求 1 秒内返回。已知客户系统脆弱,单接口并发超过 5 就可能被打挂。请给出改造方案。

参考答案要点

• 并发化:用 httpx.AsyncClient 加 asyncio.gather(或 TaskGroup)把 5 个调用并行,理论耗时降到最慢一个(约 800ms);AsyncClient 全局复用连接池,不要每请求新建
• 限流保护:对每个下游分别用 asyncio.Semaphore(5) 控制最大并发,而不是全局一刀切;信号量在连接池初始化时创建
• 超时与降级:每个调用独立 timeout(如 1.5s),超时或异常时该部分返回降级值(None 或占位标记),聚合层渲染「部分数据暂不可用」,不整体失败
• 取消传播:整体用 asyncio.wait_for 控制总时限,超时后取消未完成任务避免协程泄漏;用 TaskGroup 时注意异常即取消全部子任务的语义
• 缓存:客户画像等变化慢的数据加 TTL 缓存(进程内或 Redis),显著降低下游压力
• 观测:中间件埋点记录每个下游的耗时与失败率,给客户 IT 出调用画像,推动慢接口自身优化

来源:synthesized

q653 · 中等

客户给你一个 8GB 的工单导出 CSV(约 2000 万行、60 列),要求清洗(去重、状态过滤、字段标准化)后导入知识库构建语料。你的开发机只有 16GB 内存,直接 pd.read_csv 已经开始用 swap。请设计处理管道。

参考答案要点

• 分块处理:pd.read_csv 指定 chunksize(如 20 万行)迭代处理,每块先做列投影(usecols 只读需要的列)再过滤,结果分批写出或直接入库
• 内存优化:数值列 downcast、低基数 object 列转 category、显式指定 dtype 避免 int64/object 默认放大;评估后内存常能降一个量级
• 换工具:Polars(lazy 模式)或 DuckDB(直接对 CSV 跑 SQL)这类列式/惰性引擎可外存扫描大文件,语法简洁且快;再大规模考虑 Spark
• 去重策略:全量去重不能分块独立做——按主键哈希分桶,桶内去重;或先入库再用数据库唯一约束/窗口函数去重
• 增量与幂等:管道可重跑,记录处理水位(watermark),失败断点续跑;入库用批次事务加唯一键 upsert
• 验证对账:抽样核对清洗前后行数与关键字段分布,出数据质量报告给客户确认后再灌知识库

来源:synthesized

q842 · 困难

你要为一个 FastAPI + SQLAlchemy 服务搭建 pytest 测试体系,要求:API 级测试不占用真实端口、不依赖外部中间件;用独立测试数据库且用例间数据完全隔离;公共准备逻辑(建表/造数/客户端)可复用;校验类接口可低成本做边界用例。给出 conftest 设计与关键实现思路(不必写全代码)。

参考答案要点

• TestClient(app) 基于 httpx/starlette 传输层直接调 ASGI 应用,不占端口不起服务,是 API 级测试入口
• conftest.py 分层 fixture:session 级建测试引擎与全部表(指向 sqlite/tmp 或专用 test schema),function 级每用例开独立 Session,yield 后 rollback/清表保证隔离
• 依赖覆盖:app.dependency_overrides[get_db] = 测试 Session,让路由拿到测试库而不是生产连接;造数封装成工厂 fixture 复用
• @pytest.mark.parametrize 做表驱动边界用例(如金额<=0、超长字段、枚举非法值),断言状态码 + response_model 字段与错误明细
• 分层执行:纯函数/pydantic 校验单测最快在前,API 集成其次,少量真实外部依赖的 E2E 放最后,CI 中失败定位快

来源:[fastapi.tiangolo.com](https://fastapi.tiangolo.com/testing/)