广州市品牌网站建设平台,网站排名优化培训电话,在dw上做网站首页导航栏,禅城容桂网站制作官网文档#xff1a;https://fastapi.tiangolo.com/zh/tutorial/sql-databases/
SQL (关系型) 数据库 FastAPI不需要你使用SQL(关系型)数据库。
但是您可以使用任何您想要的关系型数据库。
这里我们将看到一个使用SQLModel的示例。
SQLModel是在SQLAlchemy和Pydantic的基础…官网文档https://fastapi.tiangolo.com/zh/tutorial/sql-databases/
SQL (关系型) 数据库¶ FastAPI不需要你使用SQL(关系型)数据库。
但是您可以使用任何您想要的关系型数据库。
这里我们将看到一个使用SQLModel的示例。
SQLModel是在SQLAlchemy和Pydantic的基础上构建的。它是由FastAPI的同一作者制作的与需要使用SQL数据库的FastAPI应用程序完美匹配。 小贴士 你可以使用任何其他你想要的SQL或NoSQL数据库库在某些情况下称为“ORM”FastAPI不会强迫你使用任何东西。 由于SQLModel基于SQLAlchemy您可以轻松使用SQLAlchemi支持的任何数据库这使得它们也受SQLModel支持您可以很容易地将其调整为任何SQLAlchemy支持的数据库如
PostgreSQLMySQLSQLiteOracleMicrosoft SQL Server等等其它数据库
在此示例中我们将使用SQLite因为它使用单个文件并且 在Python中具有集成支持。因此您可以复制此示例并按原样来运行它。
稍后对于您的产品级别的应用程序您可能会要使用像PostgreSQL这样的数据库服务器。 Tip 这儿有一个FastAPI和PostgreSQL的官方项目生成器全部基于Docker包括前端和更多工具https://github.com/tiangolo/full-stack-fastapi-postgresql 这是一个非常简单而简短的教程如果你想了解数据库、SQL或更高级的功能请参阅SQLModel文档。
安装SQLModel
首先确保创建虚拟环境激活它然后安装sqlmodel
pip install sqlmodel
Successfully installed SQLAlchemy-2.0.36 sqlmodel-0.0.22 使用单个模型创建应用程序
我们将首先使用单个SQLModel模型创建该应用程序最简单的第一个版本。
稍后我们将通过以下多种型号来提高它的安全性和多功能性。
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)secret_name: strsqlite_file_name database.db
sqlite_url fsqlite:///{sqlite_file_name}connect_args {check_same_thread: False}
engine create_engine(sqlite_url, connect_argsconnect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep Annotated[Session, Depends(get_session)]app FastAPI()app.on_event(startup)
def on_startup():create_db_and_tables()app.post(/heroes/)
def create_hero(hero: Hero, session: SessionDep) - Hero:session.add(hero)session.commit()session.refresh(hero)return heroapp.get(/heroes/)
def read_heroes(session: SessionDep,offset: int 0,limit: Annotated[int, Query(le100)] 100,
) - list[Hero]:heroes session.exec(select(Hero).offset(offset).limit(limit)).all()return heroesapp.get(/heroes/{hero_id})
def read_hero(hero_id: int, session: SessionDep) - Hero:hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)return heroapp.delete(/heroes/{hero_id})
def delete_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)session.delete(hero)session.commit()return {ok: True}
创建模型
导入SQLModel并创建数据库模型
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)secret_name: str其他版本和变体
Hero类与Pydantic模型非常相似事实上在下面它实际上是一个Pydantic模式。
存在一些差异
tableTrue告诉SQLModel这是一个表模型它应该表示SQL数据库中的一个表它不仅仅是一个数据模型就像任何其他常规Pydantic类一样。字段primary_keyTrue告诉SQLModel id是SQL数据库中的主键您可以在SQLModel文档中了解有关SQL主键的更多信息。通过将类型设置为int|NoneSQLModel将知道该列在SQL数据库中应该是INTEGER并且应该是NULLABLE。字段indexTrue告诉SQLModel它应该为该列创建SQL索引这样在读取由该列筛选的数据时可以更快地在数据库中查找。SQLModel将知道声明为str的内容将是TEXT类型的SQL列或VARCHAR具体取决于数据库。
创建引擎
SQLModel引擎其下实际上是SQLAlchemy引擎负责保存与数据库的连接。
您将有一个单一的引擎对象用于所有代码连接到同一个数据库。
sqlite_file_name database.db
sqlite_url fsqlite:///{sqlite_file_name}connect_args {check_same_thread: False}
engine create_engine(sqlite_url, connect_argsconnect_args) 使用check_same_threadFalse允许FastAPI在不同线程中使用相同的SQLite数据库。这是必要的因为一个请求可能会使用多个线程例如在依赖关系中。
别担心根据代码的结构方式我们将确保稍后每个请求使用一个SQLModel会话这实际上是check_same_thread试图实现的。
创建表格
然后我们添加一个函数该函数使用SQLModel.media.create_allengine为所有表模型创建表。
def create_db_and_tables():SQLModel.metadata.create_all(engine)
创建会话依赖关系
会话是将对象存储在内存中并跟踪数据中所需的任何更改然后使用引擎与数据库通信。
我们将使用yield创建一个FastAPI依赖关系为每个请求提供一个新的Session。这就是确保我们每个请求使用单个会话的原因。
然后我们创建一个带注释的依赖项SessionDep以简化将使用此依赖项的其余代码。 def get_session():with Session(engine) as session:yield sessionSessionDep Annotated[Session, Depends(get_session)]
启动时创建数据库表
我们将在应用程序启动时创建数据库表。
app FastAPI()app.on_event(startup)
def on_startup():create_db_and_tables()
在这里我们在应用程序启动事件上创建表。
对于生产环境您可能会使用在启动应用程序之前运行的迁移脚本。 小贴士 SQLModel将有封装Alembic的迁移实用程序但现在您可以直接使用Alembic。 创建英雄库
因为每个SQLModel模型也是一个Pydantic模型所以您可以在使用Pydantics模型的相同类型注释中使用它。
例如如果你声明一个Hero类型的参数它将从JSON正文中读取。
同样您可以将其声明为函数的返回类型然后数据的形状将显示在自动API文档UI中。
app.post(/heroes/)
def create_hero(hero: Hero, session: SessionDep) - Hero:session.add(hero)session.commit()session.refresh(hero)return hero
在这里我们使用SessionDep依赖项Session将新的Hero添加到Session实例中将更改提交到数据库中刷新Hero中的数据然后返回它。
读英雄库
我们可以使用select从数据库中读取Heros。我们可以包含一个限制和偏移量来对结果进行分页。
app.get(/heroes/)
def read_heroes(session: SessionDep,offset: int 0,limit: Annotated[int, Query(le100)] 100,
) - list[Hero]:heroes session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes
读一个英雄条目
我们可以读一个英雄。
app.get(/heroes/{hero_id})
def read_hero(hero_id: int, session: SessionDep) - Hero:hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)return hero
如果不是英雄
引发HTTPException状态码404详细信息“未找到英雄” 删除英雄
我们也可以删除英雄。
app.delete(/heroes/{hero_id})
def delete_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)session.delete(hero)session.commit()return {ok: True}
运行应用程序
您可以运行该应用程序
fastapi-dev-main.py 然后转到/docs UI您将看到FastAPI正在使用这些模型来记录API它也将使用它们来序列化和验证数据。
使用多个模型更新应用程序
现在让我们稍微重构一下这个应用程序以提高安全性和多功能性。
如果你查看之前的应用程序在UI中你可以看到到目前为止它让客户端决定要创建的英雄的id。
我们不应该让这种情况发生他们可能会覆盖我们已经在数据库中分配的id。决定id应该由后端或数据库完成而不是由客户端完成。
此外我们为英雄创建了一个secret_name但到目前为止我们到处都在返回它这不是什么秘密。。。
我们将通过添加一些额外的模型来解决这些问题。SQLModel将在这里大放异彩。✨
源代码
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)class Hero(HeroBase, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: strclass HeroUpdate(HeroBase):name: str | None Noneage: int | None Nonesecret_name: str | None Nonesqlite_file_name database.db
sqlite_url fsqlite:///{sqlite_file_name}connect_args {check_same_thread: False}
engine create_engine(sqlite_url, connect_argsconnect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep Annotated[Session, Depends(get_session)]
app FastAPI()app.on_event(startup)
def on_startup():create_db_and_tables()app.post(/heroes/, response_modelHeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_heroapp.get(/heroes/, response_modellist[HeroPublic])
def read_heroes(session: SessionDep,offset: int 0,limit: Annotated[int, Query(le100)] 100,
):heroes session.exec(select(Hero).offset(offset).limit(limit)).all()return heroesapp.get(/heroes/{hero_id}, response_modelHeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)return heroapp.patch(/heroes/{hero_id}, response_modelHeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code404, detailHero not found)hero_data hero.model_dump(exclude_unsetTrue)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_dbapp.delete(/heroes/{hero_id})
def delete_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)session.delete(hero)session.commit()return {ok: True}
创建多个模型
在SQLModel中任何具有tableTrue的模型类都是表模型。 任何没有tableTrue的模型类都是数据模型这些模型实际上只是Pydantic模型带有一些小的额外功能。 使用SQLModel我们可以使用继承来避免在所有情况下复制所有字段。 HeroBase-基类
让我们从一个HeroBase模型开始该模型包含所有模型共享的所有字段 名称 年龄
nameage
class HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue) 英雄-桌子模型 然后让我们创建Hero即实际的表模型其中包含其他模型中并不总是包含的额外字段 身份证件 秘密名称
idsecret_name
因为Hero继承自HeroBase所以它也有在HeroBase中声明的字段所以Hero的所有字段都是 身份证件 名称 年龄 秘密名称
idnameagesecret_name
class HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)class Hero(HeroBase, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)secret_name: str HeroPublic-公共数据模型 接下来我们创建一个HeroPublic模型该模型将返回给API的客户端。 它具有与HeroBase相同的字段因此不包括secret_name。 最后我们英雄的身份得到了保护 它还重新声明id:int。通过这样做我们与API客户端签订了合同这样他们就可以总是期望id在那里并且是int永远不会是None。 小贴士 让返回模型确保一个值总是可用的并且总是int而不是None对API客户端非常有用他们可以编写更简单的具有这种确定性的代码。 此外自动生成的客户端将具有更简单的接口因此与您的API通信的开发人员可以更好地使用您的API。 HeroPublic中的所有字段都与HeroBase中的相同id声明为int不是None 身份证件 名称 年龄 秘密名称
idnameagesecret_name
class HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)class Hero(HeroBase, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)secret_name: strclass HeroPublic(HeroBase):id: int HeroCreate-创建英雄的数据模型 现在我们创建一个HeroCreate模型这个模型将验证来自客户端的数据。 它具有与HeroBase相同的字段还具有secret_name。 现在当客户端创建一个新英雄时他们将发送secret_name它将存储在数据库中但这些秘密名称不会在API中返回给客户端。 小贴士 这就是你处理密码的方式。接收它们但不要在API中返回它们。 您还可以在存储密码之前对其值进行哈希运算切勿以纯文本形式存储。 HeroCreate的字段包括 名称 年龄 秘密名称
nameagesecret_name
class HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)class Hero(HeroBase, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: strHeroUpdate-更新英雄的数据模型 在之前的应用程序版本中我们没有更新英雄的方法但现在有了多个模型我们可以做到。 HeroUpdate数据模型有点特殊它具有创建新英雄所需的所有相同字段但所有字段都是可选的它们都有一个默认值。这样当你更新英雄时你可以只发送你想要更新的字段。 因为所有字段实际上都发生了变化类型现在包括None它们现在的默认值为None我们需要重新声明它们。 我们真的不需要从HeroBase继承因为我们正在重新声明所有字段。为了保持一致性我会让它继承但这不是必需的。这更多的是个人品味的问题。 HeroUpdate的字段包括 名称 年龄 秘密名称
nameagesecret_name
class HeroUpdate(HeroBase):name: str | None Noneage: int | None Nonesecret_name: str | None None 使用HeroCreate创建并返回一个HeroPublic 现在我们有了多个模型我们可以更新应用程序中使用它们的部分。 我们在请求中接收HeroCreate数据模型并从中创建Hero表模型。 这个新的表模型Hero将具有客户端发送的字段并且还将具有数据库生成的id。 然后我们返回与函数中相同的表模型Hero。但是当我们使用HeroPublic数据模型声明response_model时FastAPI将使用HeroPublic来验证和序列化数据。
app.post(/heroes/, response_modelHeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero小贴士 现在我们使用response_modelHeroPublic而不是返回类型注释-HeroPublic因为我们返回的值实际上不是HeroPublic。 如果我们声明了-HeroPublic你的编辑和linter会抱怨这是理所当然的你返回的是Hero而不是HeroPublic。 通过在response_model中声明它我们告诉FastAPI去做它的事情而不会干扰类型注释以及编辑器和其他工具的帮助。 用HeroPublic阅读英雄 我们可以像以前一样读取Heros同样我们使用response_modellist[HeroPublic]来确保数据被正确验证和序列化。
app.get(/heroes/, response_modellist[HeroPublic])
def read_heroes(session: SessionDep,offset: int 0,limit: Annotated[int, Query(le100)] 100,
):heroes session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes与HeroPublic一起阅读《一个英雄》 我们可以读一个英雄
app.get(/heroes/{hero_id}, response_modelHeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)return hero使用HeroUpdate更新英雄 我们可以更新英雄。为此我们使用HTTP PATCH操作。 在代码中我们得到一个包含客户端发送的所有数据的字典只有客户端发送的数据不包括任何仅作为默认值的值。为此我们使用exclude_unsetTrue。这是主要的伎俩。 然后我们使用hero_db.sqlmodel_updatehero_data用hero_da中的数据更新hero_db。
app.patch(/heroes/{hero_id}, response_modelHeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code404, detailHero not found)hero_data hero.model_dump(exclude_unsetTrue)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db再次删除英雄 删除英雄几乎是一样的。 我们不会满足在这个项目中重构所有内容的愿望。
app.delete(/heroes/{hero_id})
def delete_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)session.delete(hero)session.commit()return {ok: True} 再次运行应用程序 您可以再次运行该应用程序
fastapi dev main.py输出信息Uvicorn正在运行http://127.0.0.1:8000按CTRLC退出 如果你转到/docs API UI你会看到它现在已经更新并且它不会期望在创建英雄时从客户端接收id等等。 回顾 您可以使用SQLModel与SQL数据库交互并使用数据模型和表模型简化代码。 您可以在SQLModel文档中了解更多信息其中有一个关于使用SQLModel和FastAPI的较长迷你教程。
实践 安装SQLModel
首先确保创建虚拟环境激活它然后安装sqlmodel
pip install sqlmodel
源代码
存储文件到sql.py
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel):name: str Field(indexTrue)age: int | None Field(defaultNone, indexTrue)class Hero(HeroBase, tableTrue):id: int | None Field(defaultNone, primary_keyTrue)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: strclass HeroUpdate(HeroBase):name: str | None Noneage: int | None Nonesecret_name: str | None Nonesqlite_file_name database.db
sqlite_url fsqlite:///{sqlite_file_name}connect_args {check_same_thread: False}
engine create_engine(sqlite_url, connect_argsconnect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep Annotated[Session, Depends(get_session)]
app FastAPI()app.on_event(startup)
def on_startup():create_db_and_tables()app.post(/heroes/, response_modelHeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_heroapp.get(/heroes/, response_modellist[HeroPublic])
def read_heroes(session: SessionDep,offset: int 0,limit: Annotated[int, Query(le100)] 100,
):heroes session.exec(select(Hero).offset(offset).limit(limit)).all()return heroesapp.get(/heroes/{hero_id}, response_modelHeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)return heroapp.patch(/heroes/{hero_id}, response_modelHeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code404, detailHero not found)hero_data hero.model_dump(exclude_unsetTrue)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_dbapp.delete(/heroes/{hero_id})
def delete_hero(hero_id: int, session: SessionDep):hero session.get(Hero, hero_id)if not hero:raise HTTPException(status_code404, detailHero not found)session.delete(hero)session.commit()return {ok: True}
启动服务
执行命令
fastapi dev sql.py
执行后显示
INFO Importing from /Users/skywalk/work/fastapi ╭─ Python module file ─╮ │ │ │ sql.py │ │ │ ╰──────────────────────╯ INFO Importing module sql
INFO Found importable FastAPI app ╭─ Importable FastAPI app ─╮ │ │ │ from sql import app │ │ │ ╰──────────────────────────╯ INFO Using import string sql:app ╭────────── FastAPI CLI - Development mode ───────────╮ │ │ │ Serving at: http://127.0.0.1:8000 │ │ │ │ API docs: http://127.0.0.1:8000/docs │ │ │ │ Running in development mode, for production use: │ │ │ │ fastapi run │ │ │ ╰─────────────────────────────────────────────────────╯ INFO: Will watch for changes in these directories: [/Users/xxx/work/fastapi]
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)
INFO: Started reloader process [37935] using WatchFiles
INFO: Started server process [37941]
INFO: Waiting for application startup.
INFO: Application startup complete.
测试
浏览docs页面 执行curl添加指令
curl -X POST \http://127.0.0.1:8000/heroes/ \-H accept: application/json \-H Content-Type: application/json \-d {name: string,age: 0,secret_name: string
} 输出
{name:string,age:0,id:2}
证明一条信息被添加
查看一下
curl http://127.0.0.1:8000/heroes/
[{name:string,age:0,id:1},{name:string,age:0,id:2}]
果然多了一条信息。
