Глава 10. Уровень данных

Если я не ошибаюсь, Дейта сыграла комика в сериале.

Брент Спайнер (фильм «Звездный путь: Следующее поколение»)

Обзор

В этой главе мы создаем постоянный дом для данных нашего сайта, наконец-то соединяя три уровня. В нем используется реляционная база данных SQLite и представлен API базы данных Python, метко названный DB-API. Базы данных, включая пакет SQLAlchemy и нереляционные базы данных, более подробно рассматриваются в главе 14.

DB-API

Уже более 20 лет в Python существует базовое определение интерфейса реляционной базы данных, называемое DB-API: PEP 249 (https://oreil.ly/4Gp9T). Любой, кто пишет Python-драйвер для реляционной базы данных, должен как минимум включить поддержку DB-API, хотя могут быть задействованы и другие возможности. Вот основные функции DB-API.

• Создание соединения conn с базой данных с помощью функции connect().

• Создание курсора curs с помощью функции conn.cursor().

• Выполнение строки SQL stmt с помощью функции curs.execute(stmt).

Функции семейства execute...() выполняют строку SQL-оператора stmt с дополнительными параметрами, перечисленными далее:

• execute(stmt), если параметров нет;

• execute(stmt, params) с параметрами params в одной последовательности (в списке или кортеже) или словаре;

• executemany(stmt, params_seq) с несколькими группами параметров в последовательности params_seq.

Существует пять способов указания параметров, но не все они поддерживаются всеми драйверами баз данных. Если оператор stmt начинается с выражения "select * from creature where" и необходимо задать строковые параметры name или country существа, оставшаяся часть строки stmt и ее параметры будут выглядеть так, как показано в табл. 10.1.

Таблица 10.1. Указание оператора и параметров

Тип	Часть, отображающая оператор	Часть, отображающая параметры
qmark	name=? or country=?	(name, country)
numeric	name=:0 or country=:1	(name, country)
format	name=%s or country=%s	(name, country)
named	name=:name or country=:country	{"name": name, "country": country}
pyformat	name=%(name)s or country=%(country)s	{"name": name, "country": country}

Первые три принимают аргумент в виде кортежа, где порядок параметров соответствует ?, :N или %s в описании оператора. Последние два принимают словарь, в котором ключи соответствуют именам в операторе.

Таким образом, полный вызов в именованном стиле будет выглядеть так, как в примере 10.1.

Пример 10.1. Использование параметров в именованном стиле

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 необходимо выполнить итерации над возвращаемыми строками данных как над кортежами Python с помощью метода fetch:

• fetchone() возвращает один кортеж, или значение None;

• fetchall() возвращает последовательность кортежей;

• fetchmany(num) возвращает до num кортежей.

SQLite

В стандартных пакетах Python есть поддержка одной базы данных (SQLite, https://www.sqlite.org) с помощью модуля sqlite3 (https://oreil.ly/CcYtJ).

SQLite необычен — в нем нет отдельного сервера баз данных. Весь код находится в библиотеке, а хранение реализовано в одном файле. Другие базы данных работают на отдельных серверах, и клиенты общаются с ними с помощью TCP/IP, используя специальные протоколы. Задействуем SQLite в качестве первого физического хранилища данных для этого веб-сайта. В главе 14 речь пойдет о других базах данных, реляционных и нереляционных, а также о более продвинутых пакетах, таких как SQLAlchemy, и методах, подобных ORM.

Сначала необходимо определить, как структуры данных, использованные на сайте (модели), могут быть представлены в базе данных. До сих пор наши единственные модели были простыми и похожими, но не идентичными: Creature и Explorer. Они станут меняться по мере того, как мы будем придумывать, что с ними делать, и позволять данным развиваться без масштабных изменений кода.

В примере 10.2 показан голый код DB-API и SQL для создания первых таблиц и работы с ними. Он использует именованные строки аргументов (значения представляются как name), поддерживаемые пакетом sqlite3.

Пример 10.2. Создание файла data/creature.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, country, area, aka)")

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)

row = curs.fetchone()

return row_to_model(row)

def get_all(name: str) -> list[Creature]:

qry = "select * from creature"

curs.execute(qry)

rows = list(curs.fetchall())

return [row_to_model(row) for row in rows]

def create(creature: Creature):

qry = """insert into creature values

(:name, :description, :country, :area, :aka)"""

params = model_to_dict(creature)

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": creature.name}

curs.execute(qry, params)

В самом верху функция init() устанавливает соединение с sqlite3 и базой данных cryptid.db. Она хранит его в переменной conn — глобальной для модуля data/creature.py. Далее переменная curs — это курсор для итерации по данным, возвращаемым при выполнении SQL-оператора SELECT. Она также является глобальной для модуля.

Две служебные функции выполняют перевод между моделями Pydantic и DB-API:

• row_to_model() преобразует кортеж, возвращаемый функцией fetch, в объект модели;

• model_to_dict() переводит Pydantic-модель в словарь, пригодный для использования в качестве именованного параметра запроса.

Фиктивные функции CRUD, хранящиеся на каждом уровне (веб → сервис → данные), теперь будут заменены. Они применяют только обычный SQL и методы DB-API в sqlite3.

Макет

До настоящего момента данные (фиктивные) изменялись поэтапно:

• в главе 8 мы составили фиктивный список creatures в файле web/creature.py;

• в главе 8 составили фиктивный список explorers в файле web/explorer.py;

• в главе 9 перенесли подделку creatures в каталог service/creature.py;

• в главе 9 перенесли подделку explorers в каталог service/explorer.py.

Теперь данные переместились в последний раз — в файл data/creature.py. Но это уже не подделка — это настоящие живые данные, хранящиеся в файле базы данных SQLite cryptids.db. Данные о существах, опять же из-за отсутствия воображения, хранятся в SQL-таблице creature в этой базе данных.

Как только вы сохраните этот новый файл, Uvicorn должен перезапуститься из верхнего файла main.py, вызывающего web/creature.py, который вызывает файл service/creature.py, и наконец перейдет к новому файлу data/creature.py.

Заставляем все это работать

У нас есть одна небольшая проблема — этот модуль никогда не вызывает свою функцию init(), поэтому в SQLite нет переменных conn и curs, которые могли бы использовать другие функции. Это вопрос конфигурации: как предоставить информацию о базе данных при запуске? Возможны следующие варианты.

• Жесткое подключение информации о базе данных в коде, как в примере 10.2.

• Передача информации по уровням. Но это нарушило бы принцип разделения уровней — уровни веб и сервиса не должны знать о внутреннем устройстве уровня данных.

• Передача информации из другого внешнего источника:

• файла конфигурации;

• переменной окружения.

Переменная окружения проста и поддерживается такими рекомендациями, как Twelve-Factor App (https://12factor.net/config). Код может включать значение по умолчанию, если переменная окружения не определена. Этот подход можно использовать также при тестировании, чтобы создать отдельную от рабочей тестовую базу данных.

В примере 10.3 определим переменную окружения CRYPTID_SQLITE_DB и присвоим ей значение по умолчанию cryptid.db. Создайте новый файл data/init.py для нового кода инициализации базы данных, чтобы его можно было использовать и для кода исследователя.

Пример 10.3. Новый модуль инициализации базы данных data/init.py

"""Инициализация базы данных SQLite"""

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):

"""Подключение к файлу БД SQLite"""

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()

Модуль Python — это синглтон, вызываемый только один раз, несмотря на многократный импорт. Таким образом, код инициализации в файле init.py запускается всего один раз, когда происходит его первый импорт.

Наконец, измените файл data/creature.py в примере 10.4, чтобы вместо него использовать новый модуль.

• Главное, уберите строки с четвертой по восьмую.

• О, в первую очередь создайте таблицу creature!

• Все поля таблицы являются строками text SQL. Это тип столбца по умолчанию в SQLite, в отличие от большинства баз данных SQL, поэтому вам не нужно было включать text ранее, но указать явно не помешает.

• Выражение if not exists позволяет избежать разрушения таблицы после ее создания.

• Поле name служит явным первичным ключом (primary key) для этой таблицы. Если в ней будет храниться много данных исследователя, этот ключ будет необходим для быстрого поиска. Альтернативой может стать ужасное сканирование таблицы, когда код базы данных должен просмотреть каждую строку, пока не найдет совпадение с полем name.

Пример 10.4. Добавление конфигурации базы данных в файл data/creature.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,

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)

При импорте объектов conn и curs из файла init.py файлу data/creature.py больше нет необходимости импортировать сам модуль sqlite3. Если только однажды не потребуется вызвать другой метод sqlite3, не являющийся методом объекта conn или curs.

Опять же эти изменения должны указать Uvicorn перезагрузить все. С этого момента тестирование с помощью любого из описанных ранее методов (HTTPie и подобные ему или автоматические формы /docs) будет показывать сохраняемые данные. Если вы добавите существо, то оно появится в следующий раз, когда вы соберете их всех.

Сделаем то же самое для исследователей в примере 10.5.

Пример 10.5. Добавление конфигурации базы данных в файл 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)

Тестируем!

Было введено очень много кода без тестов. Все ли работает? Я бы удивился, если бы это было так. Итак, создадим несколько тестов.

Сделайте в каталоге test следующие подкаталоги:

• unit — внутри уровня;

• full — по всем уровням.

Какой тип следует написать и запустить первым? Большинство людей сначала пишут автоматизированные модульные тесты — они меньше, а всех остальных частей слоя может еще не существовать. В этой книге разработка велась сверху вниз, и сейчас мы завершаем последний слой. Кроме того, в главах 8 и 9 мы тестировали вручную с помощью HTTPie и похожих инструментов. Они помогают быстро выявить ошибки и упущения. Автоматизированные тесты гарантируют, что вы не будете повторять те же ошибки в дальнейшем. Поэтому я рекомендую:

• провести несколько тестов вручную в процессе написания кода;

• выполнить модульные тесты после исправления синтаксических ошибок Python;

• провести полное тестирование после того, как будет получен полный поток данных на всех уровнях.

Полные тесты

Они вызывают конечные веб-точки, спускающие лифт кода вниз, через сервисный уровень к уровню данных, и поднимающие обратно вверх. Иногда их называют сквозными или контрактными тестами.

Получение всех исследователей

Окунуться в тестовые воды, еще не зная, кишат ли они пираньями, сможет смелый доброволец — пример 10.6.

Пример 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 (без завершающей косой черты). В файле web/explorer.py декоратор пути для функции пути get_all() имеет вид:

@router.get("/")

Это плюс предыдущий код:

router = APIRouter(prefix = "/explorer")

означает, что функция пути get_all() предоставляет URL-адрес, содержащий выражение /explorer/.

Пример 10.7 показывает, что на одну функцию пути может приходиться более одного декоратора пути.

Пример 10.7. Добавление декоратора пути без косой черты для функции пути get_all()

@router.get("")

@router.get("/")

def get_all() -> list[Explorer]:

return service.get_all()

Протестируем оба URL-адреса в примерах 10.8 и 10.9.

Пример 10.8. Тестирование конечной точки без косой черты в конце

$ 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

[]

Пример 10.9. Тестирование конечной точки с косой чертой в конце

$ 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 предпринята подобная попытка, но с поворотом сюжета.

Пример 10.10. Создание тестового исследователя с ошибкой ввода

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

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",

"country"

"msg": "field required",

"type": "value_error.missing"

}

]

}

Я написал слово country с ошибкой, хотя моя орфография обычно безупречна. Pydantic обнаружил это на веб-уровне, вернув HTTP код состояния 422 и описание проблемы. В общем, если FastAPI возвращает код 422, велика вероятность того, что Pydantic нашел виновника сбоя. Часть выражения "loc" указывает, где произошла ошибка: поле "country" ошибочно, потому что я так плохо печатаю.

Исправьте орфографию и проведите повторный тест (пример 10.11).

Пример 10.11. Создание исследователя с исправленным значением

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

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,",

"country": "US",

"description": ""

}

На этот раз вызов возвращает код статуса 201. Он традиционно получается при создании ресурса (все коды статуса группы 2xx считаются признаком успеха, а наиболее распространен простой код 200). Ответ также содержит JSON-версию только что созданного объекта Explorer.

А теперь вернемся к начальному тесту: появится ли имя Beau в тестировании по получению всех записей исследователей? Пример 10.12 отвечает на этот животрепещущий вопрос.

Пример 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",

"country": "US",

"description": ""

}

]

Отлично!

Получите записи одного исследователя

Что произойдет, если вы попытаетесь найти Beau с помощью конечной точки для получения одной записи — Get One (пример 10.13)?

Пример 10.13. Тестирование конечной точки с инструкцией Get One

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",

"country": "US",

"description": ""

}

Я использовал кавычки, чтобы сохранить пробел между именем и фамилией. В URL-адресах вы можете применять также написание вида Beau%20Buffette. Выражение %20 означает символ пробела в шестнадцатеричном коде стандарта ASCII.

Отсутствующие и дублированные данные

До сих пор я игнорировал два основных класса ошибок:

• отсутствующие данные — если вы пытаетесь получить, изменить или удалить запись исследователя с именем, которого нет в базе данных;

• дублированные данные — если попытаетесь создать запись исследователя с одним и тем же именем более одного раза.

Что же делать, если вы запрашиваете запись несуществующего или продублированного исследователя? Пока что код слишком оптимистичен, и исключения будут всплывать из бездны.

Наш друг по имени Beau только что был добавлен в базу данных. Представьте, что его злобный клон, который носит то же имя, замышляет подменить его темной ночью, используя пример 10.14.

Пример 10.14. Ошибка дублирования — попытка создать исследователя более одного раза

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

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 и базовым Starlette. Но вот последняя строка — исключение SQLite в веб-слое! Где кушетка для обмороков?

По пятам за этим следует еще один ужас — исчезновение исследователя (пример 10.15).

Пример 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, country, description = row

^^^^^^^

TypeError: cannot unpack non-iterable NoneType object

Каков хороший способ выявить их на нижнем (данные) уровне и передать детали на верхний (веб)? Возможны следующие варианты.

• Пусть SQLite выкашливает комок волос (исключение) и разбирается с ним на веб-уровне.

Но! Это смешивает уровни, что плохо. Веб-уровень не должен ничего знать о конкретных базах данных.

• Сделайте так, чтобы все функции на сервисном уровне и уровне данных возвращали Explorer | None там, где раньше они возвращали объект Explorer. В таком случае None будет означать отказ. (Вы можете урезать это, определив OptExplorer = Explorer | None в файле model/explorer.py.)

Но! Функция могла не сработать по нескольким причинам, и вам могут понадобиться подробности. А это требует редактирования большого количества кода.

• Определите исключения для утраченных (Missing) и продублированных (Duplicate) данных, включая более детальное описание проблемы. Они будут проходить через все уровни без изменений в коде, пока функции пути веб-уровня не поймают их. Кроме того, они зависят от приложения, а не от базы данных, что позволяет сохранить неприкосновенность уровней.

Но! На самом деле мне нравится этот вариант, так что он пойдет в пример 10.16.

Пример 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.17 импортируйте исключение DB-API в файл data/init.py. SQLite вызовет его при дублировании данных.

Пример 10.17. Добавление импорта исключений SQLite в файл data/init.py

from sqlite3 import connect, IntegrityError

Импортируйте и отловите эту ошибку в примере 10.18.

Пример 10.18. Изменение data/explorer.py, чтобы получить возможность отлавливать и выбрасывать исключения

from init import (conn, curs, IntegrityError)

from model.explorer import Explorer

from error import Missing, Duplicate

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

name text primary key,

country text,

description text)""")

def row_to_model(row: tuple) -> Explorer:

name, country, description = row

return Explorer(name=name,

country=country, description=description)

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

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

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 name=:name,

country=:country,

description=:description

where name=:name_orig"""

params = model_to_dict(explorer)

params["name_orig"] = 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")

Это избавляет от необходимости объявлять, что все функции возвращают выражение Explorer | None или Optional[Explorer].

Вы указываете подсказки типов только для обычных типов возврата, но не для исключений. Поскольку исключения распространяются вверх независимо от стека вызовов до тех пор, пока кто-то их не отловит, вам не придется ничего менять на сервисном уровне. А вот новый файл web/explorer.py с обработчиками исключений и соответствующим возвратом кода состояния HTTP (пример 10.19).

Пример 10.19. Обработка Missing и Duplicate исключений в файле web/explorer.py

from fastapi import APIRouter, HTTPException

from model.explorer import Explorer

from service import explorer as service

from error 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.

Пример 10.20. Повторное тестирование конечной точки Get One с отсутствующей записью исследователя и с новым исключением Missing

$ 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).

Пример 10.21. Тестирование исправления дублирования

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

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"

}

Тестирование запросов отсутствующих данных будет применяться также к конечным точкам изменения (Modify) и удаления (Delete). Можете попробовать написать для них аналогичные тесты самостоятельно.

Модульное тестирование

Модульное тестирование работает только с уровнем данных, проверяя вызовы базы данных и синтаксис SQL. Я поместил этот раздел после полных тестов, потому что хотел, чтобы исключения Missing и Duplicate уже были определены, объяснены и закодированы в файле data/creature.py. В примере 10.22 приведен скрипт тестирования test/unit/data/test_creature.py. Вот на что стоит обратить внимание.

• Вы присваиваете переменной окружения CRYPTID_SQLITE_DATABASE значение ":memory:" до импорта init или creature из data. Это значение указывает SQLite, что необходимо работать исключительно в памяти, не захламляя существующий файл базы данных и даже не создавая файл на диске. Оно проверяется в файле data/init.py при первом импорте этого модуля.

• Фикстура под названием sample передается функциям, которым нужен объект Creature.

• Тесты выполняются по порядку. В этом случае одна и та же база данных сохраняется на протяжении всего времени выполнения, а не сбрасывается между функциями. Это необходимо для того, чтобы изменения, внесенные предыдущими функциями, сохранились. В pytest у фикстуры могут быть следующие параметры:

• область действия функции (по умолчанию) — вызывается заново перед каждой тестовой функцией;

• область действия сессии — вызывается только один раз, в самом начале.

• Некоторые тесты принудительно вызывают исключения Missing или Duplicate и проверяют возможность их отлавливания.

Итак, каждый из тестов получает совершенно новый неизменный объект Creature с именем sample (пример 10.22).

Пример 10.22. Модульные тесты для файла data/creature.py

import os

import pytest

from model.creature import Creature

from error import Missing, Duplicate

# set this before data imports below for data.init

os.environ["CRYPTID_SQLITE_DB"] = ":memory:"

from data import creature

@pytest.fixture

def sample() -> Creature:

return Creature(name="yeti", country="CN", area="Himalayas",

description="Harmless Himalayan",

aka="Abominable Snowman")

def test_create(sample):

resp = creature.create(sample)

assert resp == sample

def test_create_duplicate(sample):

with pytest.raises(Duplicate):

_ = creature.create(sample)

def test_get_one(sample):

resp = creature.get_one(sample.name)

assert resp == sample

def test_get_one_missing():

with pytest.raises(Missing):

_ = creature.get_one("boxturtle")

def test_modify(sample):

creature.area = "Sesame Street"

resp = creature.modify(sample.name, sample)

assert resp == sample

def test_modify_missing():

thing: Creature = Creature(name="snurfle", country="RU", area="",

description="some thing", aka="")

with pytest.raises(Missing):

_ = creature.modify(thing.name, thing)

def test_delete(sample):

resp = creature.delete(sample.name)

assert resp is None

def test_delete_missing(sample):

with pytest.raises(Missing):

_ = creature.delete(sample.name)

Подсказка: можете сделать собственную версию test/unit/data/test_explorer.py.

Заключение

В этой главе был представлен простой уровень обработки данных с несколькими переходами вверх и вниз по стеку уровней по мере необходимости. В главе 12 рассматриваются модульные тесты для каждого уровня, а также тесты межуровневой интеграции и полные сквозные тесты. Глава 14 посвящена более глубокому изучению баз данных и подробным примерам.