前面十篇文章，我们的商品数据都存在一个叫fake_items_db的字典里。每次重启服务，新创建的商品就全丢了。

这篇我把字典换成真正的数据库。用 SQLite 把数据存到磁盘上，重启服务后数据还在。

做完后，你创建一个商品，关掉服务再启动，调用列表接口，商品还在。

为什么要换掉内存字典

现在items.py里的数据长这样：

fake_items_db={1:ItemInDB(id=1,name="Hammer",...),2:ItemInDB(id=2,name="Screwdriver",...),}

字典放在代码文件里，有三个问题：

1. 重启就丢：字典在内存里，进程结束就没了。
2. 不方便改数据：想加一条新数据进去，要么改代码，要么写一个复杂的操作逻辑。
3. 和真实项目脱节：所有正式项目都用数据库，不是字典。

SQLite 是这次最合适的起点：它是一个文件数据库，不需要装任何数据库服务，Python 自带支持。

先装依赖

这次需要用到 SQLAlchemy，一个操作数据库的 Python 库。它会帮我们把 Python 对象和数据库表对应起来：

pip install sqlalchemy

然后在pyproject.toml里加上：

dependencies = [ "fastapi[standard]>=0.115.0", "pydantic-settings>=2.0.0", "sqlalchemy>=2.0", ]

把数据库连接抽成一个模块

新建app/database.py，三件事：连数据库、创建会话、提供get_db依赖。

fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,declarative_basefrom.configimportget_settings settings=get_settings()engine=create_engine(settings.database_url,connect_args={"check_same_thread":False},)SessionLocal=sessionmaker(autocommit=False,autoflush=False,bind=engine)Base=declarative_base()defget_db():db=SessionLocal()try:yielddbfinally:db.close()

重点看这几行：

connect_args={"check_same_thread":False}

SQLite 默认只允许创建它的线程操作数据库。FastAPI 处理请求时可能在不同线程，加这个参数告诉 SQLite 放行。

defget_db():db=SessionLocal()try:yielddbfinally:db.close()

yield是 FastAPI 依赖注入的用法。每次请求进来，FastAPI 调用get_db创建一个数据库会话，请求结束时自动close。你不用手动管理会话生命周期。

Base是所有数据库模型的父类，SQLAlchemy 用它来知道哪些类对应哪些表。

把内存数据翻译成数据库表

新建app/models.py。这里的models是数据库模型，不是 Pydantic 校验模型。

fromsqlalchemyimportColumn,Integer,String,Float,Boolean,DateTimefrom.databaseimportBaseclassItem(Base):__tablename__="items"id=Column(Integer,primary_key=True,index=True)name=Column(String,index=True)price=Column(Float)is_offer=Column(Boolean,default=False)cost_price=Column(Float)created_at=Column(DateTime)created_by=Column(String,default="system")

这里有意思的是：我们同时有app/schemas.py里的ItemCreate、ItemPublic、ItemInDB，和app/models.py里的Item。

上一篇讲模型边界时说过，请求模型、响应模型和内部数据不应该混用。数据库模型是第四种模型——它描述的是数据怎么存在表里，跟接口应该接收什么、返回什么完全独立。

schemas.py负责接口层面的校验和过滤，models.py负责告诉 SQLAlchemy 怎么建表和存数据。

让 Pydantic 能读取 SQLAlchemy 对象

之前的ItemPublic和ItemInDB只能从字典或 JSON 创建。现在它们要从数据库返回的Item对象转换，需要加一行配置：

frompydanticimportConfigDictclassItemPublic(ItemBase):id:intmodel_config=ConfigDict(from_attributes=True)classItemInDB(ItemBase):id:intcost_price:floatcreated_at:datetime created_by:strmodel_config=ConfigDict(from_attributes=True)

from_attributes=True告诉 Pydantic：你可以从对象的属性（而不是字典键）读取数据。数据库查询返回的是Item对象，Pydantic 需要这个配置才能把item.name映射成ItemPublic.name。

把操作数据库的逻辑抽出来

新建app/crud.py。CRUD 是 Create、Read、Update、Delete 的缩写，这些是操作数据最基本的能力。

fromdatetimeimportdatetime,timezonefromsqlalchemy.ormimportSessionfrom.importmodels,schemasdefget_item(db:Session,item_id:int):returndb.query(models.Item).filter(models.Item.id==item_id).first()defget_items(db:Session,q:str|None=None,limit:int=10):query=db.query(models.Item)ifq:query=query.filter(models.Item.name.ilike(f"%{q}%"))returnquery.limit(limit).all()defcreate_item(db:Session,item:schemas.ItemCreate):db_item=models.Item(name=item.name,price=item.price,is_offer=item.is_offer,cost_price=item.price*0.6,created_at=datetime.now(timezone.utc),)db.add(db_item)db.commit()db.refresh(db_item)returndb_item

三个函数做的事很直接：

• get_item：查单条，找不到返回None。
• get_items：查列表，支持用q按名称模糊搜索。
• create_item：建新商品，cost_price由服务端补上，db.refresh让对象拿到数据库自增的id。

db.commit()把操作真正写入文件。如果不调用它，数据不会持久化。

改动接口，把字典换成数据库调用

现在app/routers/items.py可以删掉那个大字典了。改成这样：

fromdatetimeimportdatetime,timezonefromfastapiimportAPIRouter,HTTPException,Dependsfromsqlalchemy.ormimportSessionfrom..importcrud,schemasfrom..databaseimportget_dbfrom..dependenciesimportget_token_header,get_common_query router=APIRouter(prefix="/items",tags=["items"])@router.get("",response_model=list[schemas.ItemPublic],summary="获取商品列表",description="支持用 q 做简单名称搜索，也可以用 limit 控制返回数量。",)deflist_items(queries:dict=Depends(get_common_query),token:str=Depends(get_token_header),db:Session=Depends(get_db),):q=queries["q"]limit=queries["limit"]returncrud.get_items(db,q=q,limit=limit)@router.get("/{item_id}",response_model=schemas.ItemPublic,summary="获取单个商品",description="根据商品 ID 查询商品。商品不存在时返回 404。",)defread_item(item_id:int,token:str=Depends(get_token_header),db:Session=Depends(get_db),):item=crud.get_item(db,item_id)ifitemisNone:raiseHTTPException(status_code=404,detail="Item not found")returnitem@router.post("",response_model=schemas.ItemPublic,summary="创建商品",description="接收创建商品需要的字段，服务端会补上 ID 和内部字段。",)defcreate_item(item:schemas.ItemCreate,token:str=Depends(get_token_header),db:Session=Depends(get_db),):returncrud.create_item(db,item)

对比之前的代码，改动其实不多：

1. 函数里多了一个db: Session = Depends(get_db)。
2. fake_items_db[id]换成了crud.get_item(db, id)。
3. 手动max(fake_items_db) + 1换成了crud.create_item(db, item)，数据库自己生成 ID。

依赖注入在这里很自然：每个接口都拿到一个数据库会话，用完自动关闭。接口函数不用管连接是怎么建立的。

启动时自动建表

最后在app/main.py里加一行，让 FastAPI 启动时自动建表：

from.importmodelsfrom.databaseimportengine models.Base.metadata.create_all(bind=engine)

放在app = FastAPI(...)之前。这样第一次启动时，SQLite 会自动在当前目录创建一个fastapi_beginner_lab.db文件，并根据models.py的Item类生成items表。

在 /docs 里试一次

启动服务：

.\.venv\Scripts\Activate.ps1 python-m pip install-e.$env:PYTHONIOENCODING ="utf-8"$env:PYTHONUTF8 ="1"fastapi dev app/main.py

打开：

http://127.0.0.1:8000/docs

先用POST /items创建几个商品：

{"name":"Monitor","price":299.99,"is_offer":false}

然后停掉服务，再启动，调用GET /items。列表里还有刚才创建的商品。

这就是数据库和内存字典的差异：数据写到了磁盘上。

数据库会话不是一个全局变量

我见过有人这么写：

db=SessionLocal()# 全局@app.get("/items")deflist_items():returndb.query(...)

这个写法在 FastAPI 里有问题。SessionLocal()创建的是共享会话，所有请求共用一个连接。读操作多了会排队，写操作可能互相覆盖。

正确的做法是用Depends(get_db)，每个请求一个独立会话。这是 FastAPI 依赖注入最常用的场景之一。

动手改一下

新增GET /items/{item_id}的支持。这篇代码里已经有了，但你可以试一件事：

1. 先调POST /items创建商品，记下返回的id。
2. 再用GET /items/{id}查这条数据，确认能查到。
3. 调GET /items/999，确认返回 404。

如果查存在的数据返回 200，查不存在的返回 404，就说明数据库查询和错误处理都接上了。

到这里，这篇的目标已经完成：

• 我们把内存字典换成了 SQLite 数据库。
• 我们写了crud.py把数据库操作从接口函数里抽出来。
• 我们确认数据在服务重启后还在。

本文代码：https://github.com/tanghaojin/fastapi-beginner-lab/tree/article-11-sql-database

下一篇解决另一个问题：不想每次改完代码都去/docs里手点，用测试来确认接口没坏。

参考资料

• FastAPI SQL Databases: https://fastapi.tiangolo.com/tutorial/sql-databases/
• SQLAlchemy 2.0 Tutorial: https://docs.sqlalchemy.org/en/20/tutorial/