FastAPI
fastapi 底层就是由协程实现的,用于构建 API,是建立在 Starlette 和 Pydantic 基础上
安装依赖
bash
pip install fastapi
pip install uvicornpython
from fastapi import FastAPI
import uvicorn
# 创建 fastapi 对象
app = FastAPI()
# 让 index 函数处理路径为斜杠的网络请求
@app.get("/") # 可以处理路径是斜杠的网络请求
def index():
return {"msg": "helloworld"}
@app.get("/item/{item_id}")
def get_one(item_id):
return {"item_id": item_id}
# uvicorn: 服务名称
# main: 模块名
# app: fastapi 对象名称
if __name__ == "__main__":
uvicorn.run(
"main:app",
host="0.0.0.0",
# port=8000
port=80, # http 协议默认端口是 80
reload=True,
)补充整理:原稿里三处会直接跑不起来的地方,已修正——
fastapi = FastAPI()与@app.get变量名不一致(装饰器会报NameError: app), 统一成app = FastAPI()uvicorn.run(app="main:app")参数写法错误:应用路径是第一个位置参数, 不是app=关键字if __name__ == "main"少了下划线,写成"__main__"才会执行另外几个实践要点:
def index()改成async def index()才能发挥协程优势;但不要在 async 函数里 写阻塞调用(如time.sleep、同步的 requests),那会把整个事件循环卡住, 这种情况用普通def,FastAPI 会自动丢进线程池/item/{item_id}建议加类型标注,FastAPI 会据此做校验和自动生成文档:python@app.get("/item/{item_id}") def get_one(item_id: int): return {"item_id": item_id}- 装依赖建议加上
pip install "uvicorn[standard]"(含性能相关的可选组件);reload=True只用于开发环境,生产必须关掉- 启动后自带交互式文档:
/docs(Swagger UI)与/redoc
SQLAlchemy
ORM 将数据库中的表结构映射为编程语言中的类 将表中的行(记录)映射为类中的对象 无需编写 SQL 语句
orm 的核心价值
- 屏蔽数据库差异
- 简化开发流程
- 提高代码可读性
- 自动处理类型转换
核心层
- Schema/Types
- SQL Expression Language
- Engine —— 维护连接池
- Connection Pooling(连接池)—— 提高效率,会创建一批数据库连接并复用
- Dialect,会对不同数据库语句差异做翻译,让上层代码以统一方式操作不同数据库
补充整理:SQLAlchemy 是两层结构,这是理解它的关键——
- Core(核心层):靠近 SQL 的一层,
Engine/Connection/Dialect/SQL Expression Language都在这。写的是「用 Python 表达的 SQL」,灵活但仍有数据库味道- ORM(对象关系映射层):在 Core 之上,把表映射成类、行映射成对象
Dialect是屏蔽差异的关键:同一句select(User).where(User.id == 1), 在 MySQL 下翻译成 MySQL 方言,在 PostgreSQL 下翻译成 PG 方言,上层代码不用改。
对象关系映射层
补充整理:这一层的映射关系是——
数据库 Python 表(table) 类(class) 行(record) 对象(instance) 列(column) 类属性(attribute) 外键 关系属性 relationship()最小可用写法(SQLAlchemy 2.0 风格):
pythonfrom sqlalchemy import create_engine, String, select from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) engine = create_engine("mysql+pymysql://user:pwd@host/db", pool_pre_ping=True) Base.metadata.create_all(engine) # 建表(通常交给迁移工具做) with Session(engine) as session: session.add(User(name="路中麋鹿")) session.commit() print(session.scalars(select(User)).all())几个易踩的坑:
- 会话要用完即关:
Session不是线程安全的,Web 应用里应该「一个请求一个 Session」, 在 FastAPI 中用依赖注入管理- 查询是惰性的:
select()只是构造语句,真正发请求在session.execute()/session.scalars()时- 别忘了 commit:
session.add()只进事务,不commit()数据不会落库pool_pre_ping=True:数据库断连后自动重连,否则可能拿到失效连接- 生产环境建表用 Alembic 做迁移,不要依赖
create_all()
核心组件
引擎(Engine) 基类(base)
补充整理:三个必须分清的东西——
- Engine:连接的「工厂」+ 连接池,全局只创建一个(在应用启动时建,整个进程复用)
- Session:一次数据库会话(事务边界),每请求一个,用完关闭
- Base(declarative base):所有模型类的父类,
Base.metadata持有全部表定义,create_all()和 Alembic 都靠它顺带一提:ORM 不是万能的。复杂报表、批量更新、需要精细控制执行计划的场景, 直接写 SQL(走 Core 层)往往更清楚也更快。
