Kawai.Focus - приложение для фокусировки внимания (часть 6)

Вступление

Всем доброго дня! В предыдущей статье 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 операций:

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

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

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

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

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

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

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

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

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

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

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