Kawai-Focus 6: продолжение работы с crud и декоратор для обработки ошибок

Вступление

Всем доброго дня! В предыдущей статье Kawai.Focus - приложение для фокусировки внимания (часть 5) было сделано:

База данных SQLite3 подключена к проекту;
Создана модель таймера;
Подключены alembic и ruff для миграций;
Созданы CRUD-функции для создания и получения таймера;
Написаны два валидатора на pydantic;
Разобраны недостатки SQLIte3.

Сегодня я напишу CRUD-операции для таймера:

update — обновление таймера;
select — получение списка таймеров (для него будет отдельная схема данных);
delete — удаление таймера.

Также я добавлю декоратор @crud_error_guard для обработки ошибок, чтобы убрать соответствующую логику из CRUD-функций.

В проекте валидаторы были заменены на схемы, поэтому потребуется немного переписать тесты в validators_tests.py, адаптировав их под новые схемы.

После этого я проверю весь новый функционал CRUD-операций, создав временную базу данных с помощью декоратора @pytest.fixture.

Заваривайте чай, доставайте вкусняшки — пора “тестировать помидор на вкус”! 🍅

Обработка ошибок

Первым делом я вынесу обработку ошибок CRUD-операций в отдельный файл kawai_focus/database/decor_errors.py и создам к нему несколько строк для ошибок. Затем добавлю ещё три CRUD-операции, в которых будут обрабатываться одни и те же ошибки.

Один из принципов Python — Don't Repeat Yourself (DRY, "Не повторяйся") — гласит, что дублирование кода является плохой практикой. Поэтому хорошим решением будет избавиться от повторяющегося кода, выделив его в декоратор для обработки ошибок.

Обработка ошибок очень сильно изменится поэтому я полностью разберу его код.

Строки ошибок

В класс ErrorMessage, который находится в kawai_focus/database/errors.py я добавил новые строки ошибок:

# Другой код
CONNECTION_ERROR = 'Подключение к базе данных не удалось!'
INTEGRITY_ERROR = 'Ошибка целостности данных (нарушение уникальности или связей)!'
OPERATIONAL_ERROR = 'Ошибка выполнения SQL-запроса (возможная блокировка или повреждение БД)!'
VALIDATION_ERROR = 'Ошибка валидации данных (несоответствие ожидаемой схеме)!'

Импорты

from typing import Any, Callable, Optional
from sqlalchemy.exc import IntegrityError, OperationalError, NoResultFound
from pydantic import ValidationError

from kawai_focus.main import Logger
from kawai_focus.utils.errors import ErrorMessage

Разбор импортов:

Any, Callable, Optional (из typing_)_ — аннотации типов для функций и переменных;
IntegrityError, OperationalError, NoResultFound (из sqlalchemy.exc_)_ — исключения SQLAlchemy для обработки ошибок базы данных;
ValidationError (из pydantic_)_ — ошибка, возникающая при валидации данных через Pydantic;
Logger (из kawai_focus.main_)_ — объект логирования для записи ошибок и событий;
ErrorMessage (из kawai_focus.utils.errors_)_ — перечисление (Enum) с текстами ошибок для структурированной обработки.

Декоратор crud_error_guard ``

def crud_error_guard(func:  Callable[..., Any]) -> Callable[..., Optional[Any]] | None:
    """Декоратор для обработки ошибок CRUD"""
    
    def wrapper(*args: Any, **kwargs: Any) -> Optional[Any] | None:
        try:
            result = func(*args, **kwargs)
            return result
        except ConnectionError as err:
            Logger.error(f'{ErrorMessage.CONNECTION_ERROR.value}: {err}')
        except IntegrityError as err:
            Logger.error(f'{ErrorMessage.INTEGRITY_ERROR.value}: {err}')
        except OperationalError as err:
            Logger.error(f'{ErrorMessage.OPERATIONAL_ERROR.value}: {err}')
        except ValidationError as err:
            Logger.error(f'{ErrorMessage.VALIDATION_ERROR.value}: {err}')
        except NoResultFound as err:
            Logger.error(f'{ErrorMessage.NO_RESULT_FOUND.value}: {err}')
    return wrapper

Декоратор crud_error_guard:

Принимает функцию CRUD (func: Callable[..., Any]);
Оборачивает её в wrapper, который отслеживает исключения.

Обработка ошибок:

ConnectionError – ошибка при подключении к БД;
IntegrityError – нарушение целостности данных;
OperationalError – сбои выполнения SQL-запросов;
ValidationError – ошибки проверки данных через Pydantic;
NoResultFound – если результат запроса отсутствует в БД.

Логирование ошибок:

Каждое исключение логируется с помощью Logger.error();
Текст ошибки берётся из ErrorMessage, чтобы код был структурированным.

Возвращаемое значение:

Если всё прошло успешно → возвращает результат func();
Если возникла ошибка → логируется и возвращается None (я не указал retirn None , так как повав в ошибку функция и так вернёт его).

Получился декоратор с аннотациями, который чётко выводит свою информацию по каждой ошибке.

Схема `TimerListModel`

TimerListModel — схема данных, которая нужна для получения списка таймеров.

Для чего потребовалась отдельная схема данных для таймеров?

На экране со спискам и таймеров не нужна другая информация кроме названия;
Ещё понадобиться id таймера для crud операции, которая получит все данные конкретного таймера.

class TimerListModel(BaseModel):
    """Модель схемы данных таймера для списка"""
    
    id: int
    title: str

Разбор кода:

id: int — id таймера;
title: str — название таймера.

CRUD операции

Сейчас мне необходимо убрать обработку ошибок из функций get_timer(), new_timer() и обернуть их декоратором @crud_error_guard. Ещё мне нужно написать три функции CRUD:

list_timers() — получение списка таймеров (select);
update_timer() — обновление данных таймера (update);
del_timer() — удаление таймера по id (delete).

Старый файл cruds.py можно найти в статье Kawai.Focus - приложение для фокусировки внимания (часть 5).

Импорты

from sqlalchemy import select, insert, update, delete
from sqlalchemy.exc import NoResultFound

from kawai_focus.schemas import TimerModel, TimerListModel
from kawai_focus.database.session import db
from kawai_focus.database.models import Timer
from kawai_focus.utils.errors import ErrorMessage
from kawai_focus.utils.decor_erors import crud_error_guard

Разбор новых импортов:

update — используется для выполнения SQL-запросов на обновление данных;
delete — позволяет удалять данные из таблицы;
NoResultFound — исключение SQLAlchemy, которое возникает, если запрос select() или get() не находит нужную запись в базе данных;
TimerModel — схема данных для модели таймера, вероятно используется для валидации и сериализации данных таймера с помощью Pydantic;
ErrorMessage — класс или модуль, содержащий предопределенные текстовые сообщения об ошибках, например, для обработки ошибок CRUD.

`get_timer()`

Функция get_timer() нужна для получения таймера по id.

Модифицированная функция get_timer():

@crud_error_guard
def get_timer(timer_id: int) -> TimerModel:
    """Функция для получения данных таймера"""
    
    with db.get_session() as session:
        query = select(
            Timer.id, 
            Timer.title,
            Timer.pomodoro_time, 
            Timer.break_time, 
            Timer.break_long_time,
            Timer.count_pomodoro
        ).where(timer_id  Timer.id)
        result = session.execute(query)
        timer = result.mappings().first()
        
        return TimerModel.model_validate(obj=timer, from_attributes=True)

Разбор нового кода:

@crud_error_guard — декоратор для обработки ошибок вместо обработки ошибок в самой функции get_timer();
return TimerModel.model_validate(obj=timer, from_attributes=True):
- преобразует timer в объект TimerModel (Pydantic-модель);
- model_validate(obj=timer, from_attributes=True) позволяет конвертировать данные из SQLAlchemy-объекта в Pydantic.

`new_timer()`

Функция new_timer() нужна для создания таймера.

Модифицированная функция new_timer():

@crud_error_guard
def new_timer(data: TimerModel) -> TimerModel:
    """Функция для создания нового таймера"""
    
    with db.get_session() as session:
        query = insert(Timer).values(**data.model_dump()).returning(Timer)
        result = session.execute(query)
        
        new_timer = result.scalar_one()
        session.commit()
        
        return TimerModel.model_validate(obj=new_timer, from_attributes=True)

Разбор нового кода:

@crud_error_guard — декоратор для обработки ошибок вместо обработки ошибок в самой функции new_timer();
-> bool — был убран из аннотации, так функция больше не возвращает булево значение, а теперь вместо этого возвращает данные с моделью TimerModel;
insert(Timer).values(**data.model_dump()).returning(Timer) — вставка данных с возвратом созданной записи;
new_timer = result.scalar_one() — извлекает новый таймер;
return TimerModel.model_validate(obj=new_timer, from_attributes=True) — преобразует ORM-объект в Pydantic-модель.

Преимущества возврата данных в new_timer():

Легко проверить создала-ли функция новый таймер;
Экономия на запросах к базе данных;
После создания можно мгновенно воспользоваться новыми данными.

`list_timers()`

list_timers() — функция получения списка таймеров:

@crud_error_guard
def list_timers() -> list[TimerListModel]:
    """Функция для получения списка таймеров"""
    
    with db.get_session() as session:
        query = select(Timer.id, Timer.title)
        result = session.execute(query)
        timers = result.mappings().fetchall()
        
        return [TimerListModel.model_validate(obj=accept, from_attributes=True) for accept in timers]

Разбор кода:

@crud_error_guard — декоратор для обработки ошибок;
list_timers() -> list[TimerListModel] — функция возвращает список таймеров в виде Pydantic-моделей.

Открытие сессии с БД:

with db.get_session() as session: — создаёт контекстную сессию для работы с базой.

Формирование SQL-запроса:

query = select(Timer.id, Timer.title) — выбирает id и title всех таймеров.

Выполнение запроса и получение данных:

result = session.execute(query) — выполняет SQL-запрос;
timers = result.mappings().fetchall() — извлекает все найденные записи.

Конвертация в Pydantic-модель:

return [TimerListModel.model_validate(obj=accept, from_attributes=True) for accept in timers] — каждая запись преобразуется в TimerListModel.

`update_timer()`

update_timer() — функция обновления данных таймера:

@crud_error_guard
def update_timer(data: TimerModel) -> TimerModel:
    """Функция для обновления таймера"""
    
    with db.get_session() as session:
        query = update(Timer).values(**data.model_dump()).where(data.id  Timer.id).returning(Timer)
        result = session.execute(query)
        
        updated_timer = result.scalar_one()
        session.commit()
        
        return TimerModel.model_validate(obj=updated_timer, from_attributes=True)

Разбор кода:

@crud_error_guard — декоратор для обработки ошибок;
update_timer(data: TimerModel) -> TimerModel — функция принимает обновлённые данные TimerModel, обновляет запись и возвращает её.

Открытие сессии с БД:

with db.get_session() as session: создаёт транзакцию для безопасной работы с базой.

Формирование SQL-запроса:

update(Timer).values(**data.model_dump()).where(data.id Timer.id).returning(Timer);
Обновляет таймер по его id и возвращает обновлённые данные.

Выполнение запроса:

result = session.execute(query) выполняет SQL-запрос.

Получение обновлённого объекта

updated_timer = result.scalar_one() извлекает обновлённую запись;
Если данных нет, вызов scalar_one() приведёт к исключению NoResultFound.

Фиксация изменений

session.commit() подтверждает обновление в базе данных.

Конвертация в TimerModel

return TimerModel.model_validate(obj=updated_timer, from_attributes=True)
Преобразует ORM-объект в Pydantic-модель.

`del_timer()`

del_timer() — функция обновления данных таймера:

@crud_error_guard
def del_timer(timer_id: int) -> None:
    """Функция для удаления таймера"""
    
    with db.get_session() as session:
        query = delete(Timer).where(timer_id  Timer.id)
        result = session.execute(query)
        session.commit()

Разбор кода:

@crud_error_guard — декоратор для обработки ошибок;
del_timer(timer_id) -> None — функция удаляет таймер по переданному timer_id и возвращает None, так как не требуется возвращать объект.

Открытие сессии с БД:

with db.get_session() as session: создаёт транзакцию для безопасного выполнения запроса.

Формирование SQL-запроса:

delete(Timer).where(timer_id Timer.id) — удаляет таймер с соответствующим id.

Выполнение запроса

result = session.execute(query) выполняет SQL-запрос на удаление.

Фиксация изменений:

session.commit() подтверждает удаление из базы данных.

Tests

Мне нужны следующие тесты:

validators_tests.py — переписать из-за изменений в коде (вместо валидаторов теперь схемы);
cruds_tests.py — тесты для CRUD операций.

Редактирование `validators_tests.py`

Поскольку в проекте больше нет валидаторов, тесты нужно адаптировать под схемы, которые их заменили. Сначала я переименовал validators_tests.py в schemas_tests.py, так как теперь это тесты схем.

Старый файл validators_tests.py можно найти в статье Kawai.Focus - приложение для фокусировки внимания (часть 4)

Импорты

import pytest
from pydantic import ValidationError

from kawai_focus.utils.errors import ErrorMessage
from kawai_focus.schemas import TimerTimeModel

Разбор новых импортов:

ValidationError — исключение из Pydantic, которое возникает при невалидных данных;
TimerTimeModel — модель из schemas, предназначенная для хранения и валидации времени таймера.

`test_valid_time()`

def test_valid_time():
    """Тест корректных данных"""
    
    timer = TimerTimeModel(hh=10, mm=30, ss=15)
    assert timer.hh  10
    assert timer.mm  30
    assert timer.ss  15

Разбор нового кода:

timer = TimerTimeModel(hh=10, mm=30, ss=15) — вместо валидатора теперь создаётся схема времени, которая принимает часы, минуты и секунды.

`test_no_time()`

def test_no_time():
    """Тест исключения ситуации когда не указано время"""
    
    with pytest.raises(ValueError) as excinfo:
        TimerTimeModel(hh=0, mm=0, ss=0)
    
    assert ErrorMessage.NO_TIME.value in str(excinfo.value)

Разбор нового кода:

TimerTimeModel(hh=0, mm=0, ss=0) — теперь используется схема вместо валидатора;
assert ErrorMessage.NO_TIME.value in str(excinfo.value) — теперь ищем подстроку в строке так как схемы pydantic по другому генерируют сообщения об ошибках и тест упадёт если использовать ``.

`test_not_int_type()`

def test_not_int_type():
    """Тест исключения для некорректного типа данных"""
    
    with pytest.raises(ValidationError) as excinfo:
        TimerTimeModel(hh=None, mm=30, ss=15)
    
    assert 'Input should be a valid integer' in str(excinfo.value)

Все новые строки кода, что тут есть были разобраны ранее.

`test_negative_time()`

def test_negative_time():
    """Тест исключения для отрицательного времени"""
    
    with pytest.raises(ValueError) as excinfo:
        TimerTimeModel(hh=-1, mm=30, ss=15)
    
    assert 'Input should be greater than or equal to 0' in str(excinfo.value)

Все новые строки кода, что тут есть были разобраны ранее.

`test_exceed_seconds_or_minutes()`

def test_exceed_seconds_or_minutes():
    """Тест исключения для секунд/минут больше 59"""
    
    with pytest.raises(ValueError) as excinfo:
        TimerTimeModel(hh=0, mm=60, ss=10)
    
    assert 'Input should be less than or equal to 59' in str(excinfo.value)

Все новые строки кода, что тут есть были разобраны ранее.

`test_exceed_hours()`

def test_exceed_hours():
    """Тест исключения для часов больше 23"""
    
    with pytest.raises(ValueError) as excinfo:
        TimerTimeModel(hh=24, mm=0, ss=0)
    
    assert 'Input should be less than or equal to 23' in str(excinfo.value)

Все новые строки кода, что тут есть были разобраны ранее.

`cruds_tests.py`

Мне потребуются следующие тесты для CRUD:

test_db() — создание временной базы данных и сессии;
test_new_timer() — создание нового таймера;
test_get_timer() — получение таймера по id;
test_list_timers() — получение списка таймеров;
test_update_timer() — обновление таймера;
test_del_timer() — удаление таймера.

Импорты

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from alembic.config import Config
from alembic import command
from pathlib import Path

from kawai_focus.schemas import TimerModel
from kawai_focus.database.session import db
from kawai_focus.database.models import Timer, Base
from kawai_focus.database.cruds import list_timers, get_timer, new_timer, update_timer, del_timer

Разбор импортов Pytest и Sqlalchemy:

import pytest — используется для написания и запуска тестов в pytest;
from sqlalchemy import create_engine — оздаёт подключение к базе данных (Engine) для работы с SQLAlchemy;
from sqlalchemy.orm import sessionmaker — позволяет создавать сессии для взаимодействия с базой.

Импорты Alembic для управления миграциями:

from alembic.config import Config — загружает конфигурацию Alembic (alembic.ini и настройки миграций);
from alembic import command — выполняет команды Alembic, такие как upgrade и downgrade.

Импорты работы с файловой системой:

from pathlib import Path — упрощает работу с путями и файлами (Path("migrations")).

Импорты из проекта kawai_focus:

from kawai_focus.schemas import TimerModel — подключает Pydantic-модель таймера (TimerModel) для валидации данных;
from kawai_focus.database.session import db — управляет соединением с базой (db.get_session() для получения сессии);
from kawai_focus.database.models import Timer, Base — импортирует ORM-модель Timer и Base, от которой наследуются модели.
from kawai_focus.database.cruds import list_timers, get_timer, new_timer, update_timer, del_timer:

Импортирует CRUD-функции для работы с таймерами:

list_timers — получение списка таймеров;
get_timer — получение одного таймера по id;
new_timer — создание нового таймера;
update_timer — обновление таймера;
del_timer — удаление таймера.

`test_db()`

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

Функции фикстуры:

Подготавливает тестовое окружение перед запуском тестов;
Создаёт временные объекты (базу данных, соединение и т.д);
Автоматически очищает ресурсы после тестов.

@pytest.fixture(scope='function')
def test_db() -> sessionmaker:
    """Создание тестовой базы данных и сессии"""
    
    engine = create_engine('sqlite:///:memory:')  # Временная БД
    Base.metadata.create_all(engine)
    
    alembic_cfg = Config('./alembic.ini')
    alembic_cfg.set_main_option('script_location', './kawai_focus/database/alembic')
    command.upgrade(alembic_cfg, 'head')
    
    session = sessionmaker(bind=engine)
    db._session_factory = session
    return db._session_factory

Разбор кода:

Создание тестовой базы данных:

create_engine('sqlite:///:memory:') — создаёт временную SQLite-базу в оперативной памяти;
Base.metadata.create_all(engine) — инициализирует все таблицы.

Настройка Alembic для миграций:

alembic_cfg = Config('./alembic.ini') — загружает конфигурацию миграций;
alembic_cfg.set_main_option('script_location', './kawai_focus/database/alembic') — путь до миграций;
command.upgrade(alembic_cfg, 'head') — применяет все доступные миграции.

Создание фабрики сессий:

sessionmaker(bind=engine) — создаёт сессию, привязанную к тестовой базе;
db._session_factory = session — заменяет фабрику сессий, чтобы тесты использовали тестовую базу.

Возвращение тестовой сессии:

return db._session_factory — позволяет использовать сессию в тестах.

`test_new_timer()`

def test_new_timer(test_db: sessionmaker):
    """Тест crud создания нового таймера"""
    
    timer_data = TimerModel(title='test_timer_1', pomodoro_time=60, break_long_time=30, break_time=6, count_pomodoro=5)
    with test_db() as session:
        timer = new_timer(data=timer_data)
    
    assert timer
    assert timer.title  'test_timer_1'
    assert timer.pomodoro_time  60
    assert timer.break_long_time  30
    assert timer.break_time  6
    assert timer.count_pomodoro  5

Разбор кода:

def test_new_timer(test_db: sessionmaker): — определяет тест для проверки CRUD-операции создания таймера и принимает фикстуру test_db, которая предоставляет тестовую сессию базы данных;
timer_data = TimerModel(...) — создаёт экземпляр Pydantic-модели TimerModel, содержащий тестовые данные таймера;

Включает параметры таймера:

title='test_timer_1' — название таймера;
pomodoro_time=60 — время "помидора" (рабочий интервал);
break_long_time=30 — время долгого перерыва;
break_time=6 — время короткого перерыва;
count_pomodoro=5 — количество циклов "помидора".

with test_db() as session::

Открывает новую тестовую сессию базы данных;
Использует sessionmaker, переданный через test_db.

timer = new_timer(data=timer_data):

Вызывает new_timer(), чтобы создать новый таймер в базе данных;
Передаёт timer_data как аргумент, создавая запись в таблице Timer.

Проверки с assert:

assert timer — проверяет, что таймер успешно создан;
assert timer.title 'test_timer_1' — проверяет корректность названия;
assert timer.pomodoro_time 60 — проверяет длительность "помидора";
assert timer.break_long_time 30 — проверяет время долгого перерыва;
assert timer.break_time 6 — проверяет время короткого перерыва;
assert timer.count_pomodoro 5 — проверяет количество циклов.

`test_get_timer()`

def test_get_timer(test_db: sessionmaker):
    """Тест crud получения таймера по id"""
    
    timer_data = TimerModel(title='test_timer_1', pomodoro_time=60, break_long_time=30, break_time=6, count_pomodoro=5)
    with test_db() as session:
        new_timer(data=timer_data)
        timer = get_timer(timer_id=1)
    
    assert timer
    assert timer.title  'test_timer_1'
    assert timer.pomodoro_time  60
    assert timer.break_long_time  30
    assert timer.break_time  6
    assert timer.count_pomodoro  5

Разбор незнакомого кода:

new_timer(data=timer_data) — создаём новый таймер (для каждого теста нужно создавать новые данные), данные которого проверит get_timer();
timer = get_timer(timer_id=1) — получаем таймер по его id.

`test_list_timers()`

def test_list_timers(test_db: sessionmaker):
    """Тест crud получения списка таймеров"""
    
    timer_data_1 = TimerModel(
        title='test_timer_1', 
        pomodoro_time=60, 
        break_long_time=30, 
        break_time=6, 
        count_pomodoro=5
    )
    
    timer_data_2 = TimerModel(
        title='test_timer_2', 
        pomodoro_time=55, 
        break_long_time=25, 
        break_time=5, 
        count_pomodoro=6
    )
    
    with test_db() as session:
        new_timer(data=timer_data_1)
        new_timer(data=timer_data_2)
        timers = list_timers()
    
    assert timers
    assert timers[0].id  1
    assert timers[0].title  'test_timer_1'
    assert timers[1].id  2
    assert timers[1].title  'test_timer_2'

Разбор незнакомого кода:

new_timer(data=timer_data_1), new_timer(data=timer_data_2) — создание двух новых таймеров, которые требуются для проверки получения списка таймеров;
timers = list_timers() — получаем список таймеров;
assert timers[0].id 1 — получаем первый таймер по его индексу далее полаем его id и сравниваем с 1 (далее строки по аналогии).

`test_update_timer()`

def test_update_timer(test_db: sessionmaker):
    """Тест crud обновления таймера"""
    
    timer_data = TimerModel(title='test_timer_1', pomodoro_time=60, break_long_time=30, break_time=6, count_pomodoro=5)
    with test_db() as session:
        new_timer(data=timer_data)
        
        new_timer_data = TimerModel(
            id=1, 
            title='test_timer_2', 
            pomodoro_time=55, 
            break_long_time=25, 
            break_time=5, 
            count_pomodoro=6
        )
        timer = update_timer(data=new_timer_data)
    
    assert timer
    assert timer.title  'test_timer_2'
    assert timer.pomodoro_time  55
    assert timer.break_long_time  25
    assert timer.break_time  5
    assert timer.count_pomodoro  6

Разбор незнакомого кода:

timer = update_timer(data=new_timer_data) — обновление таймера.

`test_del_timer()`

def test_del_timer(test_db: sessionmaker):
    """Тест crud удаления таймера"""
    
    timer_data = TimerModel(title='test_timer_1', pomodoro_time=60, break_long_time=30, break_time=6, count_pomodoro=5)
    with test_db() as session:
        new_timer(data=timer_data)
        del_timer(timer_id=1)
        timer = session.query(Timer).filter_by(title='test_timer_1').first()
    
    assert timer is None

Разбор незнакомого кода:

del_timer(timer_id=1) — удаление таймера по его id;
timer = session.query(Timer).filter_by(title='test_timer_1').first() — получение первой записи в базе данных с названием таймера 'test_timer_1';
assert timer is None — тест будет пройден, если таймер не найден.

Запуск тестов

Тесты запускаю, находясь в корне проекта.

Тест `schemas_tests`

poetry run pytest kawai_fokus/tests/schemas_tests.py

Результат тестирования:

Все шесть тестов схем прошли успешно, судя по сообщению 6 passed.

Тест `cruds_tests.py`:

poetry run pytest kawai_fokus/tests/cruds_tests.py

Результат тестирования:

Все пять тестов CRUD функций тоже прошли успешно.

Анонс на следующие статьи

Следующая моя статья выйдет 19 июня 2025 года. Я наконец напишу экран конструктора таймера и реализую переход с одного экрана на другой.

Если у вас есть мысли о том, как можно улучшить проект, пишите в комментариях — с удовольствием ознакомлюсь с вашими предложениями!

Читайте продолжение — не пропустите!

Заключение

Созданы функции CRUD операций:

update_timer() — обновление таймера;
del_timer() — удаление таймера;
list_timers() — получение списка таймеров.

Обновлены функции CRUD операций:

new_timer() — создание нового таймера;
get_timer() — получение таймера.

Написан декоратор для обработки ошибок:

crud_error_guard — обрабатывает ошибки CRUD функций;

Добавлена схема для списка моделей:

TimerListModel — схема для списка таймеров;

Написаны тесты на pytest:

test_create_timer() — создание таймера;
test_get_timer() — получение таймера по его id;
test_update_timer() — обновление данных таймера;
test_del_timer() — удаление таймера;
test_list_timers() — получение списка таймеров;
test_db() — создание базы данных и сессии;
validators_tests.py превращены в тесты для схем schemas_tests.py.

Ссылки к статье

Мои статьи Arduinum628 на Код на салфетке;
Репозиторий проекта Kawai.Focus.

Kawai-Focus 6: продолжение работы с crud и декоратор для обработки ошибок

Вступление

Обработка ошибок

Строки ошибок

Импорты

Декоратор crud_error_guard ``

Схема `TimerListModel`

CRUD операции

Импорты

`get_timer()`

`new_timer()`

`list_timers()`

`update_timer()`

`del_timer()`

Tests

Редактирование `validators_tests.py`

Импорты

`test_valid_time()`

`test_no_time()`

`test_not_int_type()`

`test_negative_time()`

`test_exceed_seconds_or_minutes()`

`test_exceed_hours()`

`cruds_tests.py`

Импорты

`test_db()`

`test_new_timer()`

`test_get_timer()`

`test_list_timers()`

`test_update_timer()`

`test_del_timer()`

Запуск тестов

Тест `schemas_tests`

Тест `cruds_tests.py`:

Анонс на следующие статьи

Заключение

Ссылки к статье

Автор

Eugene Kaddo

Комментарии

Вступление

Обработка ошибок

Строки ошибок

Импорты

Декоратор crud_error_guard ``

Схема TimerListModel

CRUD операции

Импорты

get_timer()

new_timer()

list_timers()

update_timer()

del_timer()

Tests

Редактирование validators_tests.py

Импорты

test_valid_time()

test_no_time()

test_not_int_type()

test_negative_time()

test_exceed_seconds_or_minutes()

test_exceed_hours()

cruds_tests.py

Импорты

test_db()

test_new_timer()

test_get_timer()

test_list_timers()

test_update_timer()

test_del_timer()

Запуск тестов

Тест schemas_tests

Тест cruds_tests.py:

Анонс на следующие статьи

Заключение

Ссылки к статье

Автор

Eugene Kaddo

Комментарии

Схема `TimerListModel`

`get_timer()`

`new_timer()`

`list_timers()`

`update_timer()`

`del_timer()`

Редактирование `validators_tests.py`

`test_valid_time()`

`test_no_time()`

`test_not_int_type()`

`test_negative_time()`

`test_exceed_seconds_or_minutes()`

`test_exceed_hours()`

`cruds_tests.py`

`test_db()`

`test_new_timer()`

`test_get_timer()`

`test_list_timers()`

`test_update_timer()`

`test_del_timer()`

Тест `schemas_tests`

Тест `cruds_tests.py`: