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 匹配。后两个使用一个字典，其中的键与语句中的名称相匹配。

• 使用命名式参数

stmt = """select * from creature where

name=:name or country=:country"""

params = {"name": "yeti", "country": "CN"}

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

"""Initialize SQLite database"""

import os

from pathlib import Path

from sqlite3 import connect, Connection, Cursor, IntegrityError

conn: Connection | None = None

curs: Cursor | None = None

def get_db(name: str|None = None, reset: bool = False):

"""Connect to SQLite database file"""

global conn, curs

if conn:

if not reset:

return

conn = None

if not name:

name = os.getenv("CRYPTID_SQLITE_DB")

top_dir = Path(__file__).resolve().parents[1] # repo top

db_dir = top_dir / "db"

db_name = "cryptid.db"

db_path = str(db_dir / db_name)

name = os.getenv("CRYPTID_SQLITE_DB", db_path)

conn = connect(name, check_same_thread=False)

curs = conn.cursor()

get_db()

data/creature.py

from .init import curs

from model.creature import Creature

curs.execute("""create table if not exists creature(

name text primary key,

description text,

country text,

area text,

aka text)""")

def row_to_model(row: tuple) -> Creature:

(name, description, country, area, aka) = row

return Creature(name, description, country, area, aka)

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, :country, :area, :aka)"""

params = model_to_dict(creature)

curs.execute(qry, params)

return get_one(creature.name)

def modify(creature: Creature) -> Creature:

qry = """update creature

set country=:country,

name=:name,

description=:description,

area=:area,

aka=:aka

where name=:name_orig"""

params = model_to_dict(creature)

params["name_orig"] = creature.name

_ = curs.execute(qry, params)

return get_one(creature.name)

def delete(creature: Creature) -> bool:

qry = "delete from creature where name = :name"

params = {"name": creature.name}

res = curs.execute(qry, params)

return bool(res)

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

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

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

data/explorer.py

from .init import curs

from model.explorer import Explorer

curs.execute("""create table if not exists explorer(

name text primary key,

country text,

description text)""")

def row_to_model(row: tuple) -> Explorer:

return Explorer(name=row[0], country=row[1], description=row[2])

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, country, description)

values (:name, :country, :description)"""

params = model_to_dict(explorer)

_ = curs.execute(qry, params)

return get_one(explorer.name)

def modify(name: str, explorer: Explorer) -> Explorer:

qry = """update explorer

set country=:country,

name=:name,

description=:description

where name=:name_orig"""

params = model_to_dict(explorer)

params["name_orig"] = explorer.name

_ = 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)

10.3 测试

• 启动

$ export CRYPTID_SQLITE_DB=cryptid.db

$ python main.py

INFO: Will watch for changes in these directories: ['/home/andrew/code/fastapi/example/ch8']

INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

INFO: Started reloader process [6659] using WatchFiles

INFO: Started server process [6661]

INFO: Waiting for application startup.

INFO: Application startup complete.

• 测试

$ http post localhost:8000/explorer name="Beau Buffette", contry="US"

HTTP/1.1 422 Unprocessable Entity

content-length: 174

content-type: application/json

date: Thu, 13 Jun 2024 06:11:44 GMT

server: uvicorn

{

"detail": [

{

"loc": [

"body",

"country"

],

"msg": "field required",

"type": "value_error.missing"

},

{

"loc": [

"body",

"description"

],

"msg": "field required",

"type": "value_error.missing"

}

]

}