FastAPI-10 数据层

烟波钓叟

10 数据层

本章终于为我们的网站数据创建了一个持久的家园，最终将三个层连接起来。本章使用关系数据库 SQLite，并介绍了 Python 的数据库 API（DB-API）。第14章将详细介绍数据库，包括 SQLAlchemy 软件包和非关系型数据库。

10.1 DB-API

20多年来，Python一直包含关系数据库接口的基本定义，称为 DB-API：PEP 249。任何为关系数据库编写Python驱动程序的人都应该至少包含对 DB-API 的支持，当然也可能包含其他功能。
这些是 DB-API 的主要功能：

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

execute...()函数运行带有可选参数的 SQL 语句 stmt 字符串，参数如下所示：

• execute(stmt) 如果没有参数
  
• execute(stmt,params)，参数为单个序列（列表或元组）或 dict
  
• executemany(stmt,params_seq)，在参数序列 params_seq 中包含多个参数组。
  

有五种指定参数的方式，但并非所有数据库驱动程序都支持所有方式。如果我们有一条以 "select * from creature where "开头的语句 stmt，并且我们想为生物的名称或国家指定字符串参数，那么 stmt 字符串的其余部分及其参数将如表所示。

前三项使用元组参数，参数顺序与语句中的 ?、:N 或 %s 匹配。后两个使用一个字典，其中的键与语句中的名称相匹配。

• 使用命名式参数
  

1. stmt = """select * from creature where
2.     name=:name or country=:country"""
3. params = {"name": "yeti", "country": "CN"}
4. curs.execute(stmt, params)

复制代码

对于 SQL INSERT、DELETE 和 UPDATE 语句，execute() 返回的值会告诉你它是如何工作的。对于SELECT语句，您可以使用fetch方法遍历返回的数据行（Python元组）：

• fetchone() 返回一个元组，或者返回 None。
  
• fetchall() 返回一个元组序列。
  
• fetchmany(num) 最多返回num个元组。
  

参考资料

• 软件测试精品书籍文档下载持续更新 https://github.com/china-testing/python-testing-examples 请点赞，谢谢！
  
• 本文涉及的python测试开发库 谢谢点赞！ https://github.com/china-testing/python_cn_resouce
  
• python精品书籍下载 https://github.com/china-testing/python_cn_resouce/blob/main/python_good_books.md
  
• Linux精品书籍下载 https://www.cnblogs.com/testing-/p/17438558.html
  

10.2 SQLite

Python 标准软件包中的sqlite3模块支持数据库（SQLite）。
SQLite 不同寻常：它没有单独的数据库服务器。所有代码都在一个库中，存储在一个文件中。其他数据库运行单独的服务器，客户端通过 TCP/IP 使用特定协议与之通信。
首先，我们需要定义如何在数据库中表示我们在网站中使用的数据结构（模型）。到目前为止，我们唯一的模型是简单、相似但不完全相同的： 生物和资源管理器。随着我们想出更多的方法来使用这些模型，它们也会随之改变，并在不修改大量代码的情况下让数据不断发展。
data/init.py

1. """Initialize SQLite database"""
3. import os
4. from pathlib import Path
5. from sqlite3 import connect, Connection, Cursor, IntegrityError
7. conn: Connection | None = None
8. curs: Cursor | None = None
10. def get_db(name: str|None = None, reset: bool = False):
11.     """Connect to SQLite database file"""
12.     global conn, curs
13.     if conn:
14.         if not reset:
15.             return
16.         conn = None
17.     if not name:
18.         name = os.getenv("CRYPTID_SQLITE_DB")
19.         top_dir = Path(__file__).resolve().parents[1] # repo top
20.         db_dir = top_dir / "db"
21.         db_name = "cryptid.db"
22.         db_path = str(db_dir / db_name)
23.         name = os.getenv("CRYPTID_SQLITE_DB", db_path)
24.     conn = connect(name, check_same_thread=False)
25.     curs = conn.cursor()
27. get_db()

复制代码

data/creature.py

1. from .init import curs
2. from model.creature import Creature
4. curs.execute("""create table if not exists creature(
5.                 name text primary key,
6.                 description text,
7.                 country text,
8.                 area text,
9.                 aka text)""")
11. def row_to_model(row: tuple) -> Creature:
12.     (name, description, country, area, aka) = row
13.     return Creature(name, description, country, area, aka)
15. def model_to_dict(creature: Creature) -> dict:
16.     return creature.dict()
18. def get_one(name: str) -> Creature:
19.     qry = "select * from creature where name=:name"
20.     params = {"name": name}
21.     curs.execute(qry, params)
22.     return row_to_model(curs.fetchone())
24. def get_all() -> list[Creature]:
25.     qry = "select * from creature"
26.     curs.execute(qry)
27.     return [row_to_model(row) for row in curs.fetchall()]
28.    
29. def create(creature: Creature) -> Creature:
30.     qry = """insert into creature values
31.           (:name, :description, :country, :area, :aka)"""
32.     params = model_to_dict(creature)
33.     curs.execute(qry, params)
34.     return get_one(creature.name)
36. def modify(creature: Creature) -> Creature:
37.     qry = """update creature
38.              set country=:country,
39.                  name=:name,
40.                  description=:description,
41.                  area=:area,
42.                  aka=:aka
43.              where name=:name_orig"""
44.     params = model_to_dict(creature)
45.     params["name_orig"] = creature.name
46.     _ = curs.execute(qry, params)
47.     return get_one(creature.name)
49. def delete(creature: Creature) -> bool:
50.     qry = "delete from creature where name = :name"
51.     params = {"name": creature.name}
52.     res = curs.execute(qry, params)
53.     return bool(res)

复制代码

在Pydantic模型和DB-API之间有两个实用功能：

• row_too_model() 将获取函数返回的元组转换为模型对象。
  
• model_to_dict()将 Pydantic 模型转换为字典，适合用作命名查询参数。
  

到目前为止，每一层（网络 → 服务 → 数据）都有虚假的CRUD函数，现在这些函数将被取代。它们只使用普通SQL和 sqlite3中的DB-API法。
data/explorer.py

1. from .init import curs
2. from model.explorer import Explorer
4. curs.execute("""create table if not exists explorer(
5.                 name text primary key,
6.                 country text,
7.                 description text)""")
9. def row_to_model(row: tuple) -> Explorer:
10.     return Explorer(name=row[0], country=row[1], description=row[2])
12. def model_to_dict(explorer: Explorer) -> dict:
13.     return explorer.dict() if explorer else None
15. def get_one(name: str) -> Explorer:
16.     qry = "select * from explorer where name=:name"
17.     params = {"name": name}
18.     curs.execute(qry, params)
19.     return row_to_model(curs.fetchone())
21. def get_all() -> list[Explorer]:
22.     qry = "select * from explorer"
23.     curs.execute(qry)
24.     return [row_to_model(row) for row in curs.fetchall()]
25.    
26. def create(explorer: Explorer) -> Explorer:
27.     qry = """insert into explorer (name, country, description)
28.              values (:name, :country, :description)"""
29.     params = model_to_dict(explorer)
30.     _ = curs.execute(qry, params)
31.     return get_one(explorer.name)
33. def modify(name: str, explorer: Explorer) -> Explorer:
34.     qry = """update explorer
35.              set country=:country,
36.              name=:name,
37.              description=:description
38.              where name=:name_orig"""
39.     params = model_to_dict(explorer)
40.     params["name_orig"] = explorer.name
41.     _ = curs.execute(qry, params)
42.     explorer2 = get_one(explorer.name)
43.     return explorer2
45. def delete(explorer: Explorer) -> bool:
46.     qry = "delete from explorer where name = :name"
47.     params = {"name": explorer.name}
48.     res = curs.execute(qry, params)
49.     return bool(res)

复制代码

10.3 测试

• 启动
  

1. $ export CRYPTID_SQLITE_DB=cryptid.db
2. $ python main.py
3. INFO:     Will watch for changes in these directories: ['/home/andrew/code/fastapi/example/ch8']
4. INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
5. INFO:     Started reloader process [6659] using WatchFiles
6. INFO:     Started server process [6661]
7. INFO:     Waiting for application startup.
8. INFO:     Application startup complete.

复制代码

• 测试
  

1. $ http post localhost:8000/explorer name="Beau Buffette", contry="US"
2. HTTP/1.1 422 Unprocessable Entity
3. content-length: 174
4. content-type: application/json
5. date: Thu, 13 Jun 2024 06:11:44 GMT
6. server: uvicorn
8. {
9.     "detail": [
10.         {
11.             "loc": [
12.                 "body",
13.                 "country"
14.             ],
15.             "msg": "field required",
16.             "type": "value_error.missing"
17.         },
18.         {
19.             "loc": [
20.                 "body",
21.                 "description"
22.             ],
23.             "msg": "field required",
24.             "type": "value_error.missing"
25.         }
26.     ]
27. }

复制代码

来源:https://www.cnblogs.com/testing-/p/18244574
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！