前面十篇文章我们的商品数据都存在一个叫fake_items_db的字典里。每次重启服务新创建的商品就全丢了。这篇我把字典换成真正的数据库。用 SQLite 把数据存到磁盘上重启服务后数据还在。做完后你创建一个商品关掉服务再启动调用列表接口商品还在。为什么要换掉内存字典现在items.py里的数据长这样fake_items_db{1:ItemInDB(id1,nameHammer,...),2:ItemInDB(id2,nameScrewdriver,...),}字典放在代码文件里有三个问题重启就丢字典在内存里进程结束就没了。不方便改数据想加一条新数据进去要么改代码要么写一个复杂的操作逻辑。和真实项目脱节所有正式项目都用数据库不是字典。SQLite 是这次最合适的起点它是一个文件数据库不需要装任何数据库服务Python 自带支持。先装依赖这次需要用到 SQLAlchemy一个操作数据库的 Python 库。它会帮我们把 Python 对象和数据库表对应起来pip install sqlalchemy然后在pyproject.toml里加上dependencies [ fastapi[standard]0.115.0, pydantic-settings2.0.0, sqlalchemy2.0, ]把数据库连接抽成一个模块新建app/database.py三件事连数据库、创建会话、提供get_db依赖。fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,declarative_basefrom.configimportget_settings settingsget_settings()enginecreate_engine(settings.database_url,connect_args{check_same_thread:False},)SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)Basedeclarative_base()defget_db():dbSessionLocal()try:yielddbfinally:db.close()重点看这几行connect_args{check_same_thread:False}SQLite 默认只允许创建它的线程操作数据库。FastAPI 处理请求时可能在不同线程加这个参数告诉 SQLite 放行。defget_db():dbSessionLocal()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__itemsidColumn(Integer,primary_keyTrue,indexTrue)nameColumn(String,indexTrue)priceColumn(Float)is_offerColumn(Boolean,defaultFalse)cost_priceColumn(Float)created_atColumn(DateTime)created_byColumn(String,defaultsystem)这里有意思的是我们同时有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_configConfigDict(from_attributesTrue)classItemInDB(ItemBase):id:intcost_price:floatcreated_at:datetime created_by:strmodel_configConfigDict(from_attributesTrue)from_attributesTrue告诉 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.iditem_id).first()defget_items(db:Session,q:str|NoneNone,limit:int10):querydb.query(models.Item)ifq:queryquery.filter(models.Item.name.ilike(f%{q}%))returnquery.limit(limit).all()defcreate_item(db:Session,item:schemas.ItemCreate):db_itemmodels.Item(nameitem.name,priceitem.price,is_offeritem.is_offer,cost_priceitem.price*0.6,created_atdatetime.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 routerAPIRouter(prefix/items,tags[items])router.get(,response_modellist[schemas.ItemPublic],summary获取商品列表,description支持用 q 做简单名称搜索也可以用 limit 控制返回数量。,)deflist_items(queries:dictDepends(get_common_query),token:strDepends(get_token_header),db:SessionDepends(get_db),):qqueries[q]limitqueries[limit]returncrud.get_items(db,qq,limitlimit)router.get(/{item_id},response_modelschemas.ItemPublic,summary获取单个商品,description根据商品 ID 查询商品。商品不存在时返回 404。,)defread_item(item_id:int,token:strDepends(get_token_header),db:SessionDepends(get_db),):itemcrud.get_item(db,item_id)ifitemisNone:raiseHTTPException(status_code404,detailItem not found)returnitemrouter.post(,response_modelschemas.ItemPublic,summary创建商品,description接收创建商品需要的字段服务端会补上 ID 和内部字段。,)defcreate_item(item:schemas.ItemCreate,token:strDepends(get_token_header),db:SessionDepends(get_db),):returncrud.create_item(db,item)对比之前的代码改动其实不多函数里多了一个db: Session Depends(get_db)。fake_items_db[id]换成了crud.get_item(db, id)。手动max(fake_items_db) 1换成了crud.create_item(db, item)数据库自己生成 ID。依赖注入在这里很自然每个接口都拿到一个数据库会话用完自动关闭。接口函数不用管连接是怎么建立的。启动时自动建表最后在app/main.py里加一行让 FastAPI 启动时自动建表from.importmodelsfrom.databaseimportengine models.Base.metadata.create_all(bindengine)放在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 1fastapi dev app/main.py打开http://127.0.0.1:8000/docs先用POST /items创建几个商品{name:Monitor,price:299.99,is_offer:false}然后停掉服务再启动调用GET /items。列表里还有刚才创建的商品。这就是数据库和内存字典的差异数据写到了磁盘上。数据库会话不是一个全局变量我见过有人这么写dbSessionLocal()# 全局app.get(/items)deflist_items():returndb.query(...)这个写法在 FastAPI 里有问题。SessionLocal()创建的是共享会话所有请求共用一个连接。读操作多了会排队写操作可能互相覆盖。正确的做法是用Depends(get_db)每个请求一个独立会话。这是 FastAPI 依赖注入最常用的场景之一。动手改一下新增GET /items/{item_id}的支持。这篇代码里已经有了但你可以试一件事先调POST /items创建商品记下返回的id。再用GET /items/{id}查这条数据确认能查到。调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/
)
