欢迎光临
我们一直在努力

零基础入门python46:FastAPI 配置、环境变量与依赖注入

零基础入门python46:FastAPI 配置、环境变量与依赖注入

FastAPI ?????

一、上一篇课后练习讲解

上一篇练习围绕“ASGI、应用对象与自动文档”。参考做法是先运行上一篇的测试,再用一个成功请求和一个失败请求验证边界;本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件,运行下面代码可以观察本篇要求的成功和失败状态;数据库写入全部放在明确事务边界内。
答案要点:FastAPI main 创建应用、注册路由并提供 /health;ASGI 参数错误在业务函数前返回 422,/docs 可访问。

完整答案文件:app/main.py

完整参考答案文件

完整文件:app/main.py

from fastapi import FastAPI
app = FastAPI(title="Blog Community API")
@app.get("/health")
def health():
return {"status": "ok"}

验收:运行项目测试(FastAPI/Flask 使用 python -m pytest -q,Django 使用 python manage.py test),再按本文 curl 或 Docker 命令检查预期状态码。

完整参考答案文件

下面是交付项目中真实存在的完整文件 fastapi-blog/app/main.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码;复制单个函数会丢失上下文,因此这里提供整份文件。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from . import ai, articles, auth, extras
from .database import Base, engine

Base.metadata.create_all(engine)
app = FastAPI(title='Blog Community API', version='2.0')
app.include_router(auth.router); app.include_router(articles.router); app.include_router(ai.router); app.include_router(extras.router)
@app.get('/api/health')
def health(): return {'status': 'ok'}
@app.exception_handler(ValueError)
def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(status_code=400, content={'code': 'BAD_REQUEST', 'message': str(exc)})

requirements.txt:

fastapi==0.115.6
uvicorn[standard]==0.34.0

安装并启动:

python m venv .venv
.venv\\Scripts\\python.exe m pip install r requirements.txt
.venv\\Scripts\\python.exe m uvicorn app.main:app reload

app.main:app 的含义是“导入 app/main.py 中名为 app 的对象”。–reload 只用于开发,生产环境使用固定 worker 和进程管理器。

二、本篇要解决的真实问题

把数据库 URL、JWT 密钥和调试开关放进配置层;Depends 让路由声明需要什么,而不是自己创建全局 Session。

三、请求是怎样走完整条链路的

客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任,排错时按这个顺序定位。

四、先看一个最小代码片段

@router.get("/api/health")
def health():
return {"status": "ok"}

这里的注释不是装饰:它说明数据从哪里来、为什么不能相信客户端,以及失败时系统应保持什么状态。

五、完整文件代码

app/database.py(当前阶段完整文件)

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = 'sqlite:///./blog.db'
engine = create_engine(DATABASE_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

class Base(DeclarativeBase): pass

def get_db():
db = SessionLocal()
try: yield db
finally: db.close()

阅读方法:先找路由函数,再找它的 Depends、输入 schema、数据库操作和 response_model;这五处合起来才是一个功能。

app/main.py(当前阶段完整文件)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from . import ai, articles, auth, extras
from .database import Base, engine

Base.metadata.create_all(engine)
app = FastAPI(title='Blog Community API', version='2.0')
app.include_router(auth.router); app.include_router(articles.router); app.include_router(ai.router); app.include_router(extras.router)
@app.get('/api/health')
def health(): return {'status': 'ok'}
@app.exception_handler(ValueError)
def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(status_code=400, content={'code': 'BAD_REQUEST', 'message': str(exc)})

阅读方法:先找路由函数,再找它的 Depends、输入 schema、数据库操作和 response_model;这五处合起来才是一个功能。

六、安装、启动与验收

..\\..\\.venv\\Scripts\\python.exe m pip install r requirements.txt
..\\..\\.venv\\Scripts\\python.exe m pytest q
..\\..\\.venv\\Scripts\\python.exe m uvicorn app.main:app reload

打开 /docs,按顺序完成注册、登录、创建文章、分页查询、修改删除、评论点赞。错误请求必须看到明确状态码:校验错误 422,未登录 401,无权限 403,重复点赞 409。测试应全部通过。

七、本篇验收清单

  • 能指出输入校验发生在哪个 schema。
  • 能解释 Session 何时提交、何时回滚。
  • 能用第二个用户验证资源隔离。
  • 能复现一个失败场景并说明原因。

八、课后练习

围绕“配置、环境变量与依赖注入”新增一个测试用例,写出请求、预期响应和断言;下一篇开头会给出参考实现,并继续使用本项目。

五、配置对象和依赖注入:让代码适应三种环境

数据库地址、JWT 密钥和调试开关属于配置,不应该散落在路由函数里。配置对象在启动时校验,依赖注入则让测试可以替换它:

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
app_name: str = "blog-api"
database_url: str = "sqlite:///./blog.db"
jwt_secret: str
debug: bool = False
model_config = SettingsConfigDict(env_file=".env", extra="ignore")

@lru_cache
def get_settings() > Settings:
# 只缓存配置对象,不缓存数据库连接;测试可清空缓存换环境。
return Settings()

本地 .env 示例:

JWT_SECRET=dev-secret-for-local-only
DATABASE_URL=sqlite:///./blog.db
DEBUG=true

缺少 JWT_SECRET 时应该启动失败,而不是偷偷使用公开的默认密钥。测试可以用 app.dependency_overrides[get_settings] 替换配置,结束时在 finally 中清理覆盖。这样开发、测试、生产共用同一套路由,差异只由环境提供。

验收时先用不含密钥的环境启动,记录清晰的校验错误;再恢复变量访问 health。PowerShell 中用 Get-ChildItem Env:JWT_SECRET 检查当前进程是否真的看到变量。常见事故是 .env 被提交到 Git,或在一个终端设置变量、另一个终端启动服务。

六、当前项目中的完整文件(对照阅读)

七、配置变更如何被验证

为 Settings 写三条测试:没有 JWT_SECRET 时构造对象抛 ValidationError;设置合法值后能读取 database_url;环境变量覆盖默认值。测试结束清理 os.environ,避免用例顺序影响结果。

配置项还应分级:可以公开的服务名、需要保护的密钥、可以热更新的限流阈值。密钥不应该打印到启动日志,即便 debug=True 也不能把完整 Settings 对象 repr 出来。部署时将 .env.example 提交到仓库,里面只放字段名和示例占位值。

八、补充代码文件

本篇结束:完整模块文件

本节不是代码片段,而是本篇结束时该模块的完整版本。请先备份旧文件,再整体替换;替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理,未涉及的代码先不要自行删减。

app/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = 'sqlite:///./blog.db'
engine = create_engine(DATABASE_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

class Base(DeclarativeBase): pass

def get_db():
db = SessionLocal()
try: yield db
finally: db.close()

赞(0)
未经允许不得转载:171主机测评 » 零基础入门python46:FastAPI 配置、环境变量与依赖注入
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址