本章终于为我们的网站数据创建了一个持久的家园,最终将三个层连接起来。本章使用关系数据库 SQLite,并介绍了 Python 的数据库 API(DB-API)。第14章将详细介绍数据库,包括 SQLAlchemy 软件包和非关系型数据库。
20多年来,Python一直包含关系数据库接口的基本定义,称为 DB-API:PEP 249。任何为关系数据库编写Python驱动程序的人都应该至少包含对 DB-API 的支持,当然也可能包含其他功能。
这些是 DB-API 的主要功能:
execute...()函数运行带有可选参数的 SQL 语句 stmt 字符串,参数如下所示:
有五种指定参数的方式,但并非所有数据库驱动程序都支持所有方式。如果我们有一条以 "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元组):
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之间有两个实用功能:
到目前为止,每一层(网络 → 服务 → 数据)都有虚假的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)
$ 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"
}
]
}
参与评论
手机查看
返回顶部