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 匹配。后两个使用一个字典，其中的键与语句中的名称相匹配。
[*]使用命名式参数
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 # 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:
qry = "select * from creature"
curs.execute(qry)
return
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, country=row, description=row)
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:
qry = "select * from explorer"
curs.execute(qry)
return
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 using WatchFiles
INFO: Started server process
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"
}
]
}
来源:https://www.cnblogs.com/testing-/p/18244574
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！
页:
[1]