feat: added CRUD operations for timers, created decorator for error handling, added schema ; refactor: updated functions and ; test: moved tests from to , written tests for CRUD operations on timers and database.

kawai_focus/database/cruds.py

@@ -1,57 +1,78 @@
-from sqlalchemy import select, insert
-from sqlalchemy.exc import SQLAlchemyError, OperationalError, NoResultFound
-from pydantic import ValidationError
+from sqlalchemy import select, insert, update, delete
+from sqlalchemy.exc import NoResultFound
 
-from kawai_focus.schemas import TimerModel
+from kawai_focus.schemas import TimerModel, TimerListModel
 from kawai_focus.database.session import db
 from kawai_focus.database.models import Timer
-from kawai_focus.main import Logger
+from kawai_focus.utils.errors import ErrorMessage
+from kawai_focus.database.decor_erors import crud_error_guard
 
 
+@crud_error_guard
 def get_timer(timer_id: int) -> TimerModel:
     """Функция для получения данных таймера"""
 
-    try:
-        with db.get_session() as session:
-            timer_model = Timer 
-            query = select(
-                timer_model.id, 
-                timer_model.title,
-                timer_model.pomodoro_time, 
-                timer_model.break_time, 
-                timer_model.break_long_time,
-                timer_model.count_pomodoro
-            ).where(timer_id == timer_model.id)
-            result = session.execute(query)
-            timer = result.mappings().first()
+    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)
-    except (ConnectionError, SQLAlchemyError, TimeoutError, OperationalError, ValidationError, NoResultFound) as err:
-        Logger.error(f'{err.__class__.__name__}: {err}')
+        return TimerModel.model_validate(obj=timer, from_attributes=True)
 
 
-def new_timer(data: TimerModel) -> bool | None:
+@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
+def new_timer(data: TimerModel) -> TimerModel:
     """Функция для создания нового таймера"""
 
-    try:
-        with db.get_session() as session:
-            timer_model = Timer
-            query = insert(timer_model).values(**data.model_dump())
-            session.execute(query)
-            session.commit()
-    except (ConnectionError, SQLAlchemyError, TimeoutError, OperationalError, ValidationError) as err:
-        Logger.error(f'{err.__class__.__name__}: {err}')
-    else:
-        return True
+    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)
 
 
-if __name__ == '__main__':
-    # new_timer(data=TimerValidModel(
-    #     title='test_timer_2',
-    #     pomodoro_time=50,
-    #     break_long_time=25,
-    #     break_time=5,
-    #     count_pomodoro=4
-    # ))
+@crud_error_guard
+def update_timer(data: TimerModel) -> TimerModel:
+    """Функция для обновления таймера"""
 
-    print(get_timer(timer_id=1))
+    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
+def del_timer(timer_id: int) -> None:
+    """Функция для удаления таймера"""
+
+    with db.get_session() as session:
+        query = delete(Timer).where(timer_id == Timer.id)
+        session.execute(query)
+
+        session.commit()

kawai_focus/database/decor_erors.py

@@ -0,0 +1,27 @@
+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
+
+
+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

kawai_focus/main.py

@@ -16,13 +16,13 @@ Logger.setLevel(logging.DEBUG)
 
 class KawaiFocusApp(App):
     """Класс для создания приложения"""
-    
+
     title = 'Kawai.Focus'
-    
+
     def build(self):
         # Загрузка kv файла
         Builder.load_file('kv/timer_screen.kv')
-        
+
         screen_manager = ScreenManager()
         screen_manager.add_widget(TimerScreen(name='timers_screen'))
         return screen_manager

kawai_focus/schemas.py

@@ -3,7 +3,7 @@ from kawai_focus.utils.errors import ErrorMessage
 
 
 class TimerModel(BaseModel):
-    """Модель для валидации данных таймера"""
+    """Модель схемы данных таймера"""
 
     id: int | None = None
     title: str
@@ -13,8 +13,15 @@ class TimerModel(BaseModel):
     count_pomodoro: int
 
 
+class TimerListModel(BaseModel):
+    """Модель схемы данных таймера для списка"""
+
+    id: int
+    title: str
+
+
 class TimerTimeModel(BaseModel):
-    """Модель для валидации времени таймера"""
+    """Модель схемы данных времени таймера"""
 
     hh: int = Field(0, ge=0, le=23)
     mm: int = Field(0, ge=0, le=59)

kawai_focus/tests/cruds_tests.py

@@ -0,0 +1,129 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from alembic.config import Config
+from alembic import command
+
+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.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
+
+
+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.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_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
+
+
+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
+
+
+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
+
+
+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'

kawai_focus/tests/json/timer.json

@@ -0,0 +1,5 @@
+{
+    "sound_timer": "alarm-beep.mp3",
+    "zero_time": "00:00:00",
+    "custom_time": "00:00:12"
+}

kawai_focus/tests/schemas_tests.py

@@ -1,12 +1,14 @@
 import pytest
+from pydantic import ValidationError
+
 from kawai_focus.utils.errors import ErrorMessage
-from kawai_focus.utils.validators import TimerValidator
+from kawai_focus.schemas import TimerTimeModel
 
 
 def test_valid_time():
     """Тест корректных данных"""
     
-    timer = TimerValidator(hh=10, mm=30, ss=15)
+    timer = TimerTimeModel(hh=10, mm=30, ss=15)
     assert timer.hh == 10
     assert timer.mm == 30
     assert timer.ss == 15
@@ -15,43 +17,43 @@ def test_valid_time():
 def test_not_int_type():
     """Тест исключения для некорректного типа данных"""
     
-    with pytest.raises(TypeError) as excinfo:
-        TimerValidator(hh="10", mm=30, ss=15)
-    
-    assert str(excinfo.value) == ErrorMessage.NOT_INT_TYPE.value
+    with pytest.raises(ValidationError) as excinfo:
+        TimerTimeModel(hh=None, mm=30, ss=15)
+
+    assert 'Input should be a valid integer' in str(excinfo.value)
 
 
 def test_no_time():
     """Тест исключения ситуации когда не указано время"""
     
     with pytest.raises(ValueError) as excinfo:
-        TimerValidator(hh=0, mm=0, ss=0)
-    
-    assert str(excinfo.value) == ErrorMessage.NO_TIME.value
+        TimerTimeModel(hh=0, mm=0, ss=0)
+
+    assert ErrorMessage.NO_TIME.value in str(excinfo.value)
 
 
 def test_negative_time():
     """Тест исключения для отрицательного времени"""
     
     with pytest.raises(ValueError) as excinfo:
-        TimerValidator(hh=-1, mm=30, ss=15)
-    
-    assert str(excinfo.value) == ErrorMessage.NEGATIVE_TIME.value
+        TimerTimeModel(hh=-1, mm=30, ss=15)
+
+    assert 'Input should be greater than or equal to 0' in str(excinfo.value)
 
 
 def test_exceed_seconds_or_minutes():
     """Тест исключения для секунд/минут больше 59"""
     
     with pytest.raises(ValueError) as excinfo:
-        TimerValidator(hh=0, mm=60, ss=10)
-    
-    assert str(excinfo.value) == ErrorMessage.SS_MM_BIG.value
+        TimerTimeModel(hh=0, mm=60, ss=10)
+
+    assert 'Input should be less than or equal to 59' in str(excinfo.value)
 
 
 def test_exceed_hours():
     """Тест исключения для часов больше 23"""
     
     with pytest.raises(ValueError) as excinfo:
-        TimerValidator(hh=24, mm=0, ss=0)
-    
-    assert str(excinfo.value) == ErrorMessage.HH_BIG.value
+        TimerTimeModel(hh=24, mm=0, ss=0)
+
+    assert 'Input should be less than or equal to 23' in str(excinfo.value)

kawai_focus/utils/errors.py

@@ -1,5 +1,6 @@
 from enum import Enum
 
+
 class ErrorMessage(Enum):
     NO_TIME = 'Не указано время таймера!'
     NEGATIVE_TIME = 'Время таймера не может быть отрицательным!'
@@ -10,3 +11,7 @@ class ErrorMessage(Enum):
     TYPE_DICT = 'Тип данных не является словарем!'
     NOT_INT_TYPE = 'Тип данных не является целым числом!'
     KEY_ERROR = 'Ключ не найден!'
+    CONNECTION_ERROR = 'Подключение к базе данных не удалось!'
+    INTEGRITY_ERROR = 'Ошибка целостности данных (нарушение уникальности или связей)!'
+    OPERATIONAL_ERROR = 'Ошибка выполнения SQL-запроса (возможная блокировка или повреждение БД)!'
+    VALIDATION_ERROR = 'Ошибка валидации данных (несоответствие ожидаемой схеме)!'

kawai_focus/utils/utils.py

@@ -3,13 +3,14 @@ import json
 from os import path, listdir
 
 from kawai_focus.main import Logger
-from kawai_focus.utils.errors import ErrorMessage
 from kawai_focus.schemas import TimerTimeModel
+from kawai_focus.utils.errors import ErrorMessage
 
 
 class ReadJson:
     """Класс для чтения всех JSON-файлов в директории и объединения их в словарь."""
 
+
     def __init__(self, folder_json: str):
         self.folder_json = folder_json
         self._json_data = self._read_all_json()