编程开源技术交流,分享技术与知识

网站首页 > 开源技术 正文

FastAPI教程:10 数据层(fastapi session)

wxchong 2024-07-23 21:41:46 开源技术 33 ℃ 0 评论


预览

本章最终为我们网站的数据创建了一个持久的家, 最后连接了三层。 它使用关系数据库SQLite, 并引入了Python的数据库API,恰如其分地命名为DB-API。 第17章更详细地介绍了数据库, 包括 SQLAlchemy 包和非关系包 数据库。

数据库接口

二十多年来, Python 包含了一个基本定义 一个名为 DB-API: 的关系数据库接口。 为关系数据库编写 Python 驱动程序的任何人 预计至少包括对 DB-API 的支持, 尽管可能包含其他功能。

主要的 DB-API 函数如下:

  • 使用 connect() 创建与数据库的连接。
  • 使用 conn.cursor() 创建一个游标诅咒。
  • 使用 curs.execute(stmt) 执行 SQL 字符串 stmt。

执行...() 函数运行 SQL 语句 字符串 带可选参数 (有关指定参数的多种方式,请参见下文。

  • 如果没有参数,则执行(stmt)。
  • execute(stmt, params) , 单个中 序列(列表或元组)或字典。
  • 执行多个(STMT, params_seq) ,具有多个参数组 在。

有不同的方法来指定参数, 并非所有数据库驱动程序都支持所有这些驱动程序。 对于以 “从生物哪里选择*”, 我们想为 生物的名称位置, STMT 字符串的其余部分及其参数 看起来像这样:

类型

声明部分

参数部分

祁箴

名称=?或位置=?

(姓名、位置)

数值的

名称=:0或位置=:1

(姓名、位置)

格式

名称=%s 或位置=%s

(姓名、位置)

名称=:名称或位置=:位置

{“名称”:名称,“位置”:位置}

派格式

名称=%(名称)s 或位置=%(位置)s

{“名称”:名称,“位置”:位置}

前三个采用元组参数,其中参数 顺序与 ?、:N 或语句中的 %s。 最后两个采用字典,其中键与 语句中的名称。

因此,样式的完整调用看起来像 例 10-1:

具有命名样式参数的示例。

 stmt   =   """select * from creature where 
     name=:name or location=:location""" 
 params   =   {  "name"  :   "yeti"  ,   "location"  :   "Himalayas"  } 
 curs  .  execute  (  stmt  ,   params  ) 

对于 SQL 插入、删除和更新语句, 来自 execute() 的返回值告诉你它是如何工作的。 对于选择, 循环访问返回的数据行 (作为 Python 元组) 使用获取方法:

  • fetchone() 返回一个元组,或 None 。
  • fetchall() 返回一个元组序列。
  • Fetchmany(num) 最多返回 元组。

SQLite

Python 包括对一个数据库的支持 () 模块 在其标准包中。

SQLite是不寻常的:没有单独的数据库服务器。 所有代码都在库中, 并且存储位于单个文件中。 其他数据库运行单独的服务器, 客户端通过 TCP/IP 与它们通信, 使用特定协议。 让我们使用 SQLite 作为此网站的第一个物理数据存储。 第14章将包括其他数据库,关系数据库和非关系数据库, 以及更高级的软件包,如SQLAlchemy和 像ORM这样的技术。

首先,我们需要定义我们一直在使用的数据结构 网站()可以在数据库中表示。 到目前为止,我们唯一的模型是简单和相似的, 但不完全相同: 生物与探险家 . 当我们想到更多与它们有关的事情时,它们会发生变化。 并让数据在不进行大量代码更改的情况下不断发展。

示例 10-2 显示了裸 DB-API 代码和要创建的 SQL 并使用第一个表。 它使用参数字符串(其中值 表示为 :名称 ), sqlite3 包支持。

创建文件数据/生物.py使用 sqlite3

 import   sqlite3 
 from   ..model.creature   import   Creature 

 DB_NAME   =   "cryptid.db" 
 conn   =   sqlite3  .  connect  (  DB_NAME  ) 
 curs   =   _conn  .  cursor  () 

 def   init  (): 
     curs  .  execute  (  "create table creature(name, description, location)"  ) 

 def   row_to_model  (  row  :   tuple  )   ->   Creature  : 
     return   Creature  (  name  ,   description  ,   location   =   row  ) 

 def   model_to_dict  (  creature  :   Creature  )   ->   dict  : 
     return   creature  .  dict  () 

 def   get_one  (  name  :   str  )   ->   Creature  : 
     qry   =   "select * from creature where name=:name" 
     params   =   {  "name"  :   name  } 
     curs  .  execute  (  qry  ,   params  ) 
     row   =   curs  .  fetchone  (  res  ) 
     return   row_to_model  (  row  ) 

 def   get_all  (  name  :   str  )   ->   list  [  Creature  ]: 
     qry   =   "select * from creature" 
     curs  .  execute  (  qry  ) 
     rows   =   list  (  curs  .  fetchall  ()) 
     return   rows_to_models  (  rows  ) 

 def   create  (  creature  :   Creature  ): 
     qry   =   "insert into creature values" 
           "(:name, :description, :location)" 
     params   =   model_to_dict  (  creature  ) 
     res   =   curs  .  execute  (  qry  ,   params  ) 

 def   modify  (  creature  :   Creature  ): 
     return   creature 

 def   replace  (  creature  :   Creature  ): 
     return   creature 

 def   delete  (  creature  :   Creature  ): 
     qry   =   "delete from creature where name = :name" 
     params   =   {  "name"  :   name  } 
     res   =   curs  .  execute  (  qry  ,   params  ) 

在顶部附近,init() 函数将连接到 SQLite3 和数据库伪造 。 它将其存储在变量 conn 中; 这是数据/生物.py模块中的全局。 接下来,curs 变量是用于迭代的 通过执行 SQL SELECT 语句返回的数据; 它也是模块的全局。

两个效用函数在 Pydantic 模型之间转换 和数据库-API:

  • row_to_model() 将 函数返回的元组转换为模型对象。
  • model_to_dict() 将 Pydantic 模型转换为 字典,适合用作查询参数。

假的 CRUD 功能 到目前为止,每一层都存在 (网络→服务→数据) 现在将被替换。 他们只使用普通的SQL和sqlite3中的DB-API方法。

到目前为止,这些功能是非常基本的。 它们不包括对过滤、排序、 或者分页——这一切都在第14章中出现。

布局

到目前为止,(假)数据已按步骤修改:

  • :在中制作假_creatures列表。
  • :在 中制作了假_explorers列表。
  • 将假_creatures移至
  • :将假_explorers移动到。

现在数据已经最后一次移动, 下到。 但它们不再是假的: 它们是真实的实时数据, 持久化在SQLite数据库文件中。 生物数据,再次缺乏想象力, 存储在该数据库的 SQL 表生物中。

保存此新文件后, Uvicorn应该从你的顶级 重新开始, 它叫, 它调用, 最后是这个新的。

让它发挥作用

有一个小问题: 此模块从不调用其 init() 功能 所以没有SQLite连接或curs。 其他要使用的功能。

这是一个配置问题: 如何在启动时提供数据库信息。 可能性包括:

  • 硬连线代码中的数据库信息, 如上面的代码所示。
  • 通过图层向下传递信息。 但这会违反 层的分离; Web 和服务层不应了解内部 的数据层。
  • 从其他外部源传递信息,例如:

环境变量很简单, 并得到了等建议的认可。 如果 未指定环境变量。 这种方法也可用于测试, 提供独立于生产数据库的测试数据库。

在例 10-3 中, 让我们定义一个名为 CRYPTID_SQLITE_DB ,默认值为 cryptid.db 。 创建一个名为 的新文件 新的数据库初始化代码 因此,它也可以重用于资源管理器代码。

新的数据初始化模块 data/init.py

 import   sqlite3 
 import   os 

 _dbname   =   os  .  environ  .  get  (  "CRYPTID_SQLITE_DB"  ,   "cryptid.db"  ) 
 conn   =   sqlite3  .  connect  (  db_name  ) 
 curs   =   conn  .  cursor  () 

Python 模块是一个, 尽管多次导入,但只调用一次。 所以, 中的初始化代码只是 在首次导入时运行一次。

最后,修改如例 10-4 中 使用此新模块 相反:

  • 主要是删除第 4 行到第 8 行。
  • 哦,并在 第一名!
  • 表字段都是 SQL 文本字符串。 这是 SQLite 中的默认列类型 (与大多数SQL数据库不同), 所以我不需要 更早地包含文本,但 明确无妨。
  • 如果不存在,则避免破坏表 创建后。
  • 名称字段是此表的显式主键。 如果此表包含大量资源管理器数据, 该键对于快速查找是必需的。 另一种选择是可怕的, 数据库代码需要查看每一行,直到 它找到名称 的匹配项。

将数据库配置添加到数据/生物.py

 from   .init   import   conn  ,   curs 
 from   ..model.creature   import   Creature 

 curs  .  execute  (  """create table if not exists creature( 
                 name text primary key, 
                 description text, 
                 location text)"""  ) 

 def   row_to_model  (  row  :   tuple  )   ->   Creature  : 
     return   Creature  (  name  ,   description  ,   location   =   row  ) 

 def   model_to_dict  (  creature  :   Creature  )   ->   dict  : 
     return   creature  .  dict  () 

 def   get_one  (  name  :   str  )   ->   Creature  : 
     qry   =   "select * from creature where name=:name" 
     params   =   {  "name"  :   name  } 
     curs  .  execute  (  qry  ,   params  ) 
     return   row_to_model  (  curs  .  fetchone  ()) 

 def   get_all  ()   ->   list  [  Creature  ]: 
     qry   =   "select * from creature" 
     curs  .  execute  (  qry  ) 
     return   [  row_to_model  (  row  )   for   row   in   curs  .  fetchall  ()] 

 def   create  (  creature  :   Creature  )   ->   Creature  : 
     qry   =   "insert into creature values" 
           "(:name, :description, :location)" 
     params   =   model_to_dict  (  creature  ) 
     curs  .  execute  (  qry  ,   params  ) 
     return   get_one  (  creature  .  name  ) 

 def   modify  (  creature  :   Creature  )   ->   Creature  : 
     qry   =   """update creature 
              set location=:location, 
                  name=:name, 
                  description=:description 
              where name=:name0""" 
     params   =   model_to_dict  (  creature  ) 
     params  [  "name0"  ]   =   creature  .  name 
     res   =   curs  .  execute  (  qry  ,   params  ) 
     return   get_one  (  creature  .  name  ) 

 def   delete  (  creature  :   Creature  )   ->   bool  : 
     qry   =   "delete from creature where name = :name" 
     params   =   {  "name"  :   name  } 
     res   =   curs  .  execute  (  qry  ,   params  ) 
     return   bool  (  res  ) 

通过从 导入 conn 和 curs, 导入 sqlite3 本身 — 除非有一天有必要调用另一个 sqlite3 不是 conn 或 curs 对象的方法。

同样,这些变化应该会让Uvicorn重新加载 万事。 从现在开始, 使用到目前为止您见过的任何方法进行测试 (Httpie 和朋友,或自动 /docs 表单) 将显示保留的数据。 如果添加一个生物, 下次你得到所有这些时,它会在那里。

让我们对示例 10-5 中的资源管理器执行相同的操作。

将数据库配置添加到数据/资源管理器.py

 from   .init   import   conn  ,   curs 
 from   ..model.explorer   import   Explorer 

 curs  .  execute  (  """create table if not exists explorer( 
                 name text primary key, 
                 nationality text)"""  ) 

 def   row_to_model  (  row  :   tuple  )   ->   Explorer  : 
     return   Explorer  (  name  =  row  [  0  ],   nationality  =  row  [  1  ]) 

 def   model_to_dict  (  explorer  :   Explorer  )   ->   dict  : 
     return   explorer  .  dict  ()   if   explorer   else   None 

 def   get_one  (  name  :   str  )   ->   Explorer  : 
     qry   =   "select * from explorer where name=:name" 
     params   =   {  "name"  :   name  } 
     curs  .  execute  (  qry  ,   params  ) 
     return   row_to_model  (  curs  .  fetchone  ()) 

 def   get_all  ()   ->   list  [  Explorer  ]: 
     qry   =   "select * from explorer" 
     curs  .  execute  (  qry  ) 
     return   [  row_to_model  (  row  )   for   row   in   curs  .  fetchall  ()] 

 def   create  (  explorer  :   Explorer  )   ->   Explorer  : 
     qry   =   """insert into explorer (name, nationality) 
              values (:name, :nationality)""" 
     params   =   model_to_dict  (  explorer  ) 
     res   =   curs  .  execute  (  qry  ,   params  ) 
     return   get_one  (  explorer  .  name  ) 

 def   modify  (  name  :   str  ,   explorer  :   Explorer  )   ->   Explorer  : 
     qry   =   """update explorer 
              set nationality=:nationality, name=:name 
              where name=:name0""" 
     params   =   model_to_dict  (  explorer  ) 
     params  [  "name0"  ]   =   explorer  .  name 
     res   =   curs  .  execute  (  qry  ,   params  ) 
     explorer2   =   get_one  (  explorer  .  name  ) 
     return   explorer2 

 def   delete  (  explorer  :   Explorer  )   ->   bool  : 
     qry   =   "delete from explorer where name = :name" 
     params   =   {  "name"  :   explorer  .  name  } 
     res   =   curs  .  execute  (  qry  ,   params  ) 
     return   bool  (  res  ) 

测试!

这是很多没有测试的代码。 一切正常吗? 如果这一切都发生了,我真的会感到惊讶。 因此,让我们设置一些测试:

创建这些子目录 在目录下: * :在图层内 * :跨所有层

应该首先编写和运行哪种类型? 大多数人首先编写自动化单元测试; 它们更小,所有其他层块可能还不存在。 在这本书中,发展是自上而下的, 我们现在正在完成最后一层。 此外,我们进行了手动测试(与Httpie和朋友一起) 在“Web ”和“服务”章节中。 这些有助于快速暴露错误和遗漏; 自动测试确保您不会继续犯同样的错误 后。 所以,我建议:

  • 首次编写代码时进行一些手动测试。
  • 修复 Python 语法错误后进行单元测试。
  • 在所有层之间拥有完整的数据流后进行全面测试。

全面测试

这些调用 Web 终结点, 通过服务到数据将代码电梯向下, 然后再次备份。 有时这些被称为或测试。

获取所有浏览器

将脚趾浸入测试水域, 还不知道他们是否感染了食人鱼, 是勇敢的志愿者示例10-6。

测试获取所有资源管理器

$ http localhost:8000/explorer
HTTP/1.1 405 Method Not Allowed
allow: POST
content-length: 31
content-type: application/json
date: Mon, 27 Feb 2023 20:05:18 GMT
server: uvicorn

{
    "detail": "Method Not Allowed"
}

哎呀!发生了什么事?

哦。 我要求 /explorer ,而不是 /explorer/ , 并且没有 GET 方法路径函数 的 URL /explorer(没有最后的斜杠)。 在 中,路径装饰器 get_all() 路径函数为:

@router.get("/")

再加上前面的代码:

router = APIRouter(prefix = "/explorer")

表示此 get_all() 路径函数 提供包含 /explorer/ 的 URL。

例 20-7 愉快地表明您可以拥有多个 每个路径函数的路径修饰器。

为 get_all() 路径函数添加非斜杠路径修饰器

 @router  .  get  (  ""  ) 
 @router  .  get  (  "/"  ) 
 def   get_all  ()   ->   list  [  Explorer  ]: 
     return   service  .  get_all  () 

使用示例 10-8 和 10-9 中的两个 URL 进行测试:

测试非斜杠终结点。

$ http localhost:8000/explorer
HTTP/1.1 200 OK
content-length: 2
content-type: application/json
date: Mon, 27 Feb 2023 20:12:44 GMT
server: uvicorn

[]

测试斜杠终结点

$ http localhost:8000/explorer/
HTTP/1.1 200 OK
content-length: 2
content-type: application/json
date: Mon, 27 Feb 2023 20:14:39 GMT
server: uvicorn

[]

现在这两个都工作了, 创建一个资源管理器, ,然后重试“获取全部”测试。 示例 10-10 将尝试此操作, 但有一个情节转折:

测试资源管理器创建,但输入错误

$ http post localhost:8000/explorer name="Beau Buffette", natinality="United States"
HTTP/1.1 422 Unprocessable Entity
content-length: 95
content-type: application/json
date: Mon, 27 Feb 2023 20:17:45 GMT
server: uvicorn

{
    "detail": [
        {
            "loc": [
                "body",
                "nationality"
            ],
            "msg": "field required",
            "type": "value_error.missing"
        }
    ]
}

我拼错了国籍, 虽然我的拼写通常是无可挑剔的。 Pydantic在Web层中抓住了这一点, 返回 422 HTTP 状态代码和问题说明。 通常,如果 FastAPI 返回 422,则很有可能 皮丹蒂奇指了指肇事者。 “loc”部分表示错误发生的位置: 缺少“国籍”字段, 因为我是一个无能的打字员。

修复示例 10-11 中的拼写并重新测试:

使用更正值创建

$ http post localhost:8000/explorer name="Beau Buffette" nationality="United States"
HTTP/1.1 201 Created
content-length: 55
content-type: application/json
date: Mon, 27 Feb 2023 20:20:49 GMT
server: uvicorn

{
    "name": "Beau Buffette,",
    "nationality": "United States"
}

这次它返回了一个 201 状态代码, 创建资源时是传统的 (所有 状态代码都被视为表示成功, 普通 200 是最通用的)。 响应还包含 JSON 版本的 刚刚创建的资源管理器对象。

现在回到最初的测试:博会出现在 让所有探索者测试? 示例 10-12 回答了这个紧迫的问题:

最新的 create() 工作了吗?

$ http localhost:8000/explorer
HTTP/1.1 200 OK
content-length: 57
content-type: application/json
date: Mon, 27 Feb 2023 20:26:26 GMT
server: uvicorn

[
    {
        "name": "Beau Buffette",
        "nationality": "United States"
    }
]

耶。

获取一个资源管理器

现在,如果您尝试查找Beau会发生什么 获取终结点 (示例 10-13)?

测试获取一个终结点

$ http localhost:8000/explorer/"Beau Buffette"
HTTP/1.1 200 OK
content-length: 55
content-type: application/json
date: Mon, 27 Feb 2023 20:28:48 GMT
server: uvicorn

{
    "name": "Beau Buffette,",
    "nationality": "United States"
}

我使用引号来保留第一个和 姓氏。 在网址中,您还可以使用 美女%20巴菲特 ; %20 是 ASCII 中空格字符的十六进制代码。

缺失和重复的数据

到目前为止,我忽略了两个主要的错误类:

  • :如果您尝试获取、修改、 或者按不在数据库中的名称删除资源管理器。
  • :如果您尝试创建 多次具有相同名称的资源管理器。

那么,如果我要求一个不存在或重复的资源管理器怎么办? 到目前为止,代码过于乐观,例外情况将 从深渊冒泡。

我们的朋友Beau刚刚被添加到数据库中。 想象他邪恶的克隆人 (谁分享他的名字) 阴谋在某个黑夜取代他, 使用示例 10-14:

重复错误:尝试多次创建资源管理器

$ http post localhost:8000/explorer name="Beau Buffette" nationality="United States"
HTTP/1.1 500 Internal Server Error
content-length: 3127
content-type: text/plain; charset=utf-8
date: Mon, 27 Feb 2023 21:04:09 GMT
server: uvicorn

Traceback (most recent call last):
  File ".../starlette/middleware/errors.py", line 162, in call
... (lots of confusing innards here) ...
  File ".../service/explorer.py", line 11, in create
    return data.create(explorer)
           ^^^^^^^
  File ".../data/explorer.py", line 37, in create
    curs.execute(qry, params)
sqlite3.IntegrityError: UNIQUE constraint failed: explorer.name

我省略了该错误跟踪中的大部分行 (并用省略号替换了一些部分), 因为它主要包含 FastAPI 进行的内部调用 和底层的斯塔莱特。

但最后一行: Web 层中的 SQLite 异常! 昏厥的沙发在哪里?

紧随其后的是, 示例 10-15 中的另一个恐怖:一个失踪的探险家。

获取不存在的资源管理器

$ http localhost:8000/explorer/"Beau Buffalo"
HTTP/1.1 500 Internal Server Error
content-length: 3282
content-type: text/plain; charset=utf-8
date: Mon, 27 Feb 2023 21:09:37 GMT
server: uvicorn

Traceback (most recent call last):
  File ".../starlette/middleware/errors.py", line 162, in call
... (many lines of ancient cuneiform) ...
  File ".../data/explorer.py", line 11, in row_to_model
    name, nationality = row
    ^^^^^^^
TypeError: cannot unpack non-iterable NoneType object

在底部(数据)层捕获这些的好方法是什么, 并将细节传达给顶部(网络)? 可能性包括:

  • 让SQLite咳出一个毛球(例外)并处理 它在 Web 层中。
    • 但是:这会混合图层,这很糟糕。网络图层 应该不知道任何关于特定数据库的信息。
  • 在服务和数据层中创建每个函数 返回资源管理器 |没有他们过去返回资源管理器的地方。 然后,“无”表示失败。 (您可以通过定义 OptExplorer = Explorer |没有 在.)
    • 但是:它可能由于多个原因而失败,您可能会 想要详细信息。这需要大量的代码编辑。
  • 定义缺失和重复数据的例外,包括 问题的详细信息。这些将流经层 在 Web 路径函数捕获它们之前,不会更改任何代码。 它们也是特定于应用程序的,而不是特定于数据库的, 保持层的神圣性。
    • 但是:实际上,我喜欢这个,si 它在示例 10-16 中。

定义新的顶级 errors.py

 class   Missing  (  Exception  ): 
     def   __init__  (  self  ,   msg  :  str  ): 
         self  .  msg   =   msg 

 class   Duplicate  (  Exception  ): 
     def   __init__  (  self  ,   msg  :  str  ): 
         self  .  msg   =   msg 

其中每个都有一个 msg 字符串属性,可以 通知更高级别的代码发生了什么。

为了实现这一点, 在示例 10-7 中, 有导入SQLite会引发的DB-API异常 重复项:

将 SQLite 异常导入添加到 data/init 中.py

 from   sqlite3   import   connect  ,   IntegrityError 

导入并捕获示例 10-18 中的此错误:

修改数据/资源管理器.py以捕获和引发这些异常

 from   .init   import   (  conn  ,   curs  ,   IntegrityError  ) 
 from   ..model.explorer   import   Explorer 
 from   ..errors   import   Missing  ,   Duplicate 

 curs  .  execute  (  """create table if not exists explorer( 
                 name text primary key, 
                 nationality text)"""  ) 

 def   row_to_model  (  row  :   tuple  )   ->   Explorer  : 
     name  ,   nationality   =   row 
     return   Explorer  (  name  =  name  ,   nationality  =  nationality  ) 

 def   model_to_dict  (  explorer  :   Explorer  )   ->   dict  : 
     return   explorer  .  dict  () 

 def   get_one  (  name  :   str  )   ->   Explorer  : 
     qry   =   "select * from explorer where name=:name" 
     params   =   {  "name"  :   name  } 
     curs  .  execute  (  qry  ,   params  ) 
     row   =   curs  .  fetchone  () 
     if   row  : 
         return   row_to_model  (  row  ) 
     else  : 
         raise   Missing  (  msg  =  f  "Explorer   {  name  }   not found"  ) 

 def   get_all  ()   ->   list  [  Explorer  ]: 
     qry   =   "select * from explorer" 
     curs  .  execute  (  qry  ) 
     return   [  row_to_model  (  row  )   for   row   in   curs  .  fetchall  ()] 

 def   create  (  explorer  :   Explorer  )   ->   Explorer  : 
     if   not   explorer  :   return   None 
     qry   =   """insert into explorer (name, nationality) values 
              (:name, :nationality)""" 
     params   =   model_to_dict  (  explorer  ) 
     try  : 
         curs  .  execute  (  qry  ,   params  ) 
     except   IntegrityError  : 
         raise   Duplicate  (  msg  = 
             f  "Explorer   {  explorer  .  name  }   already exists"  ) 
     return   get_one  (  explorer  .  name  ) 

 def   modify  (  name  :   str  ,   explorer  :   Explorer  )   ->   Explorer  : 
     if   not   (  name   and   explorer  ):   return   None 
     qry   =   """update explorer 
              set nationality=:nationality, name=:name 
              where name=:name0""" 
     params   =   model_to_dict  (  explorer  ) 
     params  [  "name0"  ]   =   explorer  .  name 
     curs  .  execute  (  qry  ,   params  ) 
     if   curs  .  rowcount   ==   1  : 
         return   get_one  (  explorer  .  name  ) 
     else  : 
         raise   Missing  (  msg  =  f  "Explorer   {  name  }   not found"  ) 

 def   delete  (  name  :   str  ): 
     if   not   name  :   return   False 
     qry   =   "delete from explorer where name = :name" 
     params   =   {  "name"  :   name  } 
     curs  .  execute  (  qry  ,   params  ) 
     if   curs  .  rowcount   !=   1  : 
         raise   Missing  (  msg  =  f  "Explorer   {  name  }   not found"  ) 

这样就不再需要声明任何函数返回 探索者 |无或可选[资源管理器] 。 您只指示普通返回类型的类型提示, 不例外。 因为异常独立于调用堆栈向上流动 直到有人抓住他们, 这一次,我不必更改服务层中的任何内容。 但这是新的 在示例 10-19 中, 具有异常处理程序和适当的 HTTP 状态代码返回:

在 Web/资源管理器中处理缺失和重复的异常.py

 from   fastapi   import   APIRouter  ,   HTTPException 
 from   ..model.explorer   import   Explorer 
 from   ..service   import   explorer   as   service 
 from   ..errors   import   Duplicate  ,   Missing 

 router   =   APIRouter  (  prefix   =   "/explorer"  ) 

 @router  .  get  (  ""  ) 
 @router  .  get  (  "/"  ) 
 def   get_all  ()   ->   list  [  Explorer  ]: 
     return   service  .  get_all  () 

 @router  .  get  (  "/  {name}  "  ) 
 def   get_one  (  name  )   ->   Explorer  : 
     try  : 
         return   service  .  get_one  (  name  ) 
     except   Missing   as   exc  : 
         raise   HTTPException  (  status_code  =  404  ,   detail  =  exc  .  msg  ) 

 @router  .  post  (  ""  ,   status_code  =  201  ) 
 @router  .  post  (  "/"  ,   status_code  =  201  ) 
 def   create  (  explorer  :   Explorer  )   ->   Explorer  : 
     try  : 
         return   service  .  create  (  explorer  ) 
     except   Duplicate   as   exc  : 
         raise   HTTPException  (  status_code  =  404  ,   detail  =  exc  .  msg  ) 

 @router  .  patch  (  "/"  ) 
 def   modify  (  name  :   str  ,   explorer  :   Explorer  )   ->   Explorer  : 
     try  : 
         return   service  .  modify  (  name  ,   explorer  ) 
     except   Missing   as   exc  : 
         raise   HTTPException  (  status_code  =  404  ,   detail  =  exc  .  msg  ) 

 @router  .  delete  (  "/  {name}  "  ,   status_code  =  204  ) 
 def   delete  (  name  :   str  ): 
     try  : 
         return   service  .  delete  (  name  ) 
     except   Missing   as   exc  : 
         raise   HTTPException  (  status_code  =  404  ,   detail  =  exc  .  msg  ) 

测试示例 10-20 中的这些更改:

再次测试获取不存在的资源管理器之一,但缺少新的异常

$ http localhost:8000/explorer/"Beau Buffalo"
HTTP/1.1 404 Not Found
content-length: 44
content-type: application/json
date: Mon, 27 Feb 2023 21:11:27 GMT
server: uvicorn

{
    "detail": "Explorer Beau Buffalo not found"
}

好。 现在,在示例 10-21 中再次尝试邪恶克隆尝试:

测试重复修复

$ http post localhost:8000/explorer name="Beau Buffette" nationality="United States"
HTTP/1.1 404 Not Found
content-length: 50
content-type: application/json
date: Mon, 27 Feb 2023 21:14:00 GMT
server: uvicorn

{
    "detail": "Explorer Beau Buffette already exists"
}

缺少的检查也适用于修改和删除 端点。 您可以尝试为它们编写类似的测试。

单元测试

(待办事项:在丢失/重复内容后将此部分向上移动)

这些仅处理数据层, 检查数据库调用和 SQL 语法。 我已经把这个部分放在完整的测试之后 因为我想拥有 已经缺少和重复的异常 定义 解释 并编码成。 示例 10-6 列出了测试脚本 test/。 需要注意的一些事项:

  • 我设置了环境变量CRYPTID_SQLITE_DATABASE 到 “:memory:” 导入 init 或 creature 从数据.此值使 SQLite 完全在内存中工作, 不踩踏任何现有的数据库文件, 甚至在磁盘上创建一个文件。 首次导入该模块时.py它会在 中签入。
  • 将名为 sample 的传递给函数 需要一个生物对象。
  • 测试按顺序运行。在这种情况下,同一个数据库 始终保持熬夜,而不是在 功能。原因是允许对以前的更改 要保留的函数。使用 pytest, 夹具可以具有:
    • 范围(默认):每次测试前都会重新调用 功能。
    • 范围:仅在开始时调用一次。
  • 某些测试会强制出现“缺失”或“重复”异常, 并验证他们是否抓住了他们。

因此,下面的每个测试都得到了一个全新的,不变的 名为示例的生物对象。

数据/生物的单元测试.py

 import   os 
 import   pytest 
 from   ....model.creature   import   Creature 
 from   ....error   import   Missing  ,   Duplicate 

 # Set this before data.init import below 
 os  .  environ  [  "CRYPTID_SQLITE_DB"  ]   =   ":memory:" 
 from   ...data   import   init  ,   creature 

 @pytest  .  fixture 
 def   sample  ()   ->   Creature  : 
     return   Creature  (  name  =  "yeti"  , 
         description  =  "Abominable Snowman"  , 
         location  =  "Himalayas"  ) 

 def   test_create  (  sample  ): 
     resp   =   creature  .  create  (  sample  ) 
     assert   resp   ==   sample 

 def   test_create_duplicate  (  sample  ): 
     with   pytest  .  raises  (  Duplicate  ): 
         resp   =   creature  .  create  (  sample  ) 

 def   test_get_exists  (  sample  ): 
     resp   =   creature  .  get_one  (  sample  .  name  ) 
     assert   resp   ==   sample 

 def   test_get_missing  (): 
     with   pytest  .  raises  (  Missing  ): 
         resp   =   creature  .  get_one  (  "boxturtle"  ) 

 def   test_modify  (  sample  ): 
     creature  .  location   =   "Sesame Street" 
     resp   =   creature  .  modify  (  sample  .  name  ,   sample  ) 
     assert   resp   ==   sample 

 def   test_modify_missing  (): 
     bob  :   Creature   =   Creature  (  name  =  "bob"  , 
         description  =  "some guy"  ,   location  =  "somewhere"  ) 
     with   pytest  .  raises  (  Missing  ): 
         resp   =   creature  .  modify  (  bob  .  name  ,   bob  ) 

 def   test_delete  (  sample  ): 
     resp   =   creature  .  delete  (  sample  .  name  ) 
     assert   resp   is   None 

 def   test_delete_missing  (  sample  ): 
     with   pytest  .  raises  (  Missing  ): 
         resp   =   creature  .  delete  (  sample  .  name  ) 

提示:您可以制作自己的版本。

回顾

本章介绍了 简单数据处理层, 在层堆栈上上下下几次 根据需要。 第12章 包含每一层的单元测试, 以及跨层集成 和完整的端到端测试。 第17章将介绍更多的数据库深度和详细示例。

Tags:

本文暂时没有评论,来添加一个吧(●'◡'●)

欢迎 发表评论:

最近发表
标签列表