add Validate decorators.

Author: flask-pro

.github/workflows/testing.yaml

@@ -26,6 +26,7 @@ jobs:
           3.10
           3.11
           3.12
+          3.13
     - run: make test
   tox:
     runs-on: ubuntu-latest
@@ -38,6 +39,7 @@ jobs:
           3.10
           3.11
           3.12
+          3.13
     - run: make tox
   build:
     runs-on: ubuntu-latest

CHANGES.md

@@ -1,3 +1,7 @@
+## Version 4.0.0
+
+* Add `Validation` decorators.
+
 ## Version 3.0.0
 
 * Add method `paginate()` to `ReadMixin`.

Makefile

@@ -38,6 +38,7 @@ build: clean venv
 
 install: build
 	$(PIP) install dist/db_first-*.tar.gz
+	$(PRE_COMMIT) install
 
 upload_to_testpypi: build
 	$(PYTHON_VENV) -m twine upload --repository-url https://test.pypi.org/legacy/ dist/*

pyproject.toml

@@ -25,11 +25,12 @@ license = {file = "LICENSE"}
 name = "DB-First"
 readme = "README.md"
 requires-python = ">=3.9"
-version = "3.0.0"
+version = "4.0.0"
 
 [project.optional-dependencies]
 dev = [
   "build==1.3.0",
+  "mypy",
   "pre-commit==4.3.0",
   "pytest==8.4.2",
   "pytest-cov==7.0.0",

src/db_first/base.py

@@ -1,8 +1,6 @@
 from typing import Any
 from typing import Optional
 
-from sqlalchemy.engine import Result
-
 from .exc import MetaNotFound
 from .exc import OptionNotFound
 
@@ -24,42 +22,3 @@ def _get_option_from_meta(cls, name: str, default: Optional[Any] = ...) -> Any:
                 option = default
 
         return option
-
-    @classmethod
-    def deserialize_data(cls, schema_name: str, data: dict) -> dict:
-        schema = cls._get_option_from_meta(schema_name)
-        return schema().load(data)
-
-    @classmethod
-    def _clean_data(cls, data: Any) -> Any:
-        """Clearing hierarchical structures from empty values.
-
-        Cleaning occurs for objects of the list and dict types, other types do not clean.
-
-        :param data: an object for cleaning.
-        :return: cleaned object.
-        """
-
-        empty_values = ('', None, ..., [], {}, (), set())
-
-        if isinstance(data, dict):
-            cleaned_dict = {k: cls._clean_data(v) for k, v in data.items()}
-            return {k: v for k, v in cleaned_dict.items() if v not in empty_values}
-
-        elif isinstance(data, list):
-            cleaned_list = [cls._clean_data(item) for item in data]
-            return [item for item in cleaned_list if item not in empty_values]
-
-        else:
-            return data
-
-    @classmethod
-    def serialize_data(cls, schema_name: str, data: Result, fields: list = None) -> dict:
-        output_schema = cls._get_option_from_meta(schema_name)
-
-        if isinstance(data, list):
-            serialized_data = output_schema(many=True, only=fields).dump(data)
-        else:
-            serialized_data = output_schema(only=fields).dump(data)
-
-        return cls._clean_data(serialized_data)

src/db_first/decorators/__init__.py

@@ -0,0 +1,3 @@
+from .validation import Validation
+
+__all__ = ['Validation']

src/db_first/decorators/validation.py

@@ -0,0 +1,46 @@
+from collections.abc import Callable
+from collections.abc import Iterable
+from functools import wraps
+from typing import Any
+
+from marshmallow.schema import SchemaMeta
+
+
+class Validation:
+    @classmethod
+    def input(
+        cls, schema: SchemaMeta, deserialize: bool = True, keys: Iterable[str] | None = None
+    ) -> Callable:
+        def decorator(func: Callable) -> Callable:
+            @wraps(func)
+            def wrapper(self, **data) -> Any:
+                if deserialize:
+                    deserialized_data = schema(only=keys).load(data)
+                    return func(self, **deserialized_data)
+                else:
+                    schema(only=keys).validate(data)
+                    return func(self, **data)
+
+            return wrapper
+
+        return decorator
+
+    @classmethod
+    def output(
+        cls, schema: SchemaMeta, serialize: bool = False, keys: Iterable[str] | None = None
+    ) -> Callable:
+        def decorator(func: Callable) -> Callable:
+            @wraps(func)
+            def wrapper(self, *args, **kwargs) -> Any:
+                obj = func(self, *args, **kwargs)
+
+                if serialize:
+                    serialized_data = schema(only=keys).dump(obj)
+                    return serialized_data
+                else:
+                    schema(only=keys).validate(obj)
+                    return obj
+
+            return wrapper
+
+        return decorator

src/db_first/mixins/crud.py

@@ -43,19 +43,6 @@ def create_object(self, **kwargs) -> Result:
         session.commit()
         return new_obj
 
-    def create(
-        self, data: dict, deserialize: bool = False, serialize: bool = False
-    ) -> Result or dict:
-        if deserialize:
-            data = self.deserialize_data('input_schema_of_create', data)
-
-        new_object = self.create_object(**data)
-
-        if serialize:
-            return self.serialize_data('output_schema_of_create', new_object)
-
-        return new_object
-
 
 class ReadMixin:
     """Read objects from database.
@@ -95,7 +82,7 @@ def _calculate_items_per_page(
         self, session: Session, statement: Select, per_page: int
     ) -> tuple[int, int]:
         total = session.execute(
-            statement.with_only_columns(func.count()).order_by(None)
+            statement.with_only_columns(func.count(self.Meta.model.id)).order_by(None)
         ).scalar_one()
 
         if per_page == 0:
@@ -115,9 +102,7 @@ def _paginate(
         page: int = 1,
         per_page: Optional[int] = 20,
         max_per_page: Optional[int] = 100,
-        serialize: bool = False,
         include_metadata: bool = False,
-        fields: Optional[list] = None,
     ) -> dict:
         session: Session = self._get_option_from_meta('session')
 
@@ -139,30 +124,21 @@ def _paginate(
         else:
             paginated_rows = []
 
-        if serialize:
-            items['items'] = self.serialize_data('output_schema_of_read', paginated_rows, fields)
-        else:
-            items['items'] = paginated_rows
+        items['items'] = paginated_rows
 
         return items
 
-    def paginate(
+    def base_paginate(
         self,
         page: int = 1,
         per_page: Optional[int] = None,
         max_per_page: Optional[int] = None,
         statement: Optional[Select] = None,
-        deserialize: bool = False,
-        serialize: bool = False,
         include_metadata: bool = False,
-        fields: Optional[list] = None,
         **kwargs,
     ) -> Result or dict:
         model = self._get_option_from_meta('model')
 
-        if deserialize:
-            kwargs = self.deserialize_data('input_schema_of_read', kwargs)
-
         filterable_fields = self._get_option_from_meta('filterable', ())
         interval_filterable_fields = self._get_option_from_meta('interval_filterable', ())
         searchable_fields = self._get_option_from_meta('searchable', ())
@@ -189,9 +165,7 @@ def paginate(
             page=page,
             per_page=per_page,
             max_per_page=max_per_page,
-            serialize=serialize,
             include_metadata=include_metadata,
-            fields=fields,
         )
 
         return items
@@ -204,19 +178,6 @@ def read_object(self, id: Any) -> Result:
         stmt = select(model).where(model.id == id)
         return session.scalars(stmt).one()
 
-    def read(
-        self, data: dict, deserialize: bool = False, serialize: bool = False
-    ) -> Result or dict:
-        if deserialize:
-            data = self.deserialize_data('input_schema_of_read', data)
-
-        object_ = self.read_object(**data)
-
-        if serialize:
-            return self.serialize_data('output_schema_of_read', object_)
-
-        return object_
-
 
 class UpdateMixin:
     """Update object in database.
@@ -247,19 +208,6 @@ def update_object(self, id: Any, **kwargs) -> Result:
         obj = session.scalars(stmt).one()
         return obj
 
-    def update(
-        self, data: dict, deserialize: bool = False, serialize: bool = False
-    ) -> Result or dict:
-        if deserialize:
-            data = self.deserialize_data('input_schema_of_update', data)
-
-        updated_object = self.update_object(**data)
-
-        if serialize:
-            return self.serialize_data('output_schema_of_update', updated_object)
-
-        return updated_object
-
 
 class DeleteMixin:
     """Delete object from database."""
@@ -271,8 +219,3 @@ def delete_object(self, id: Any) -> None:
         model = self._get_option_from_meta('model')
 
         session.execute(delete(model).where(model.id == id))
-
-    def delete(self, data: dict, deserialize: bool = False) -> None:
-        if deserialize:
-            data = self.deserialize_data('input_schema_of_read', data)
-        self.delete_object(**data)

src/db_first/schemas/__init__.py

@@ -0,0 +1,3 @@
+from .base import BaseSchema
+
+__all__ = ['BaseSchema']

src/db_first/schemas/base.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+from marshmallow import post_dump
+from marshmallow import Schema
+
+
+class BaseSchema(Schema):
+    __empty_values__ = ('', None, ..., [], {}, (), set())
+    __skipped_keys__ = ()
+
+    @post_dump()
+    def _delete_keys_with_empty_value(
+        self, data, many=False
+    ) -> dict[Any, dict] | list[dict] | dict | list:
+        """Clearing hierarchical structures from empty values.
+
+        Cleaning occurs for objects of the list and dict types, other types do not clean.
+
+        :param data: an object for cleaning.
+        :param many: Should be set to `True` if ``obj`` is a collection so that the object will
+         be serialized to a list.
+        :return: cleaned object.
+        """
+
+        if isinstance(data, dict):
+            pre_cleaned_dict = {
+                k: self._delete_keys_with_empty_value(v, many=many) for k, v in data.items()
+            }
+
+            cleaned_dict = {}
+            for k, v in pre_cleaned_dict.items():
+                if k not in self.__skipped_keys__ and v in self.__empty_values__:
+                    continue
+                else:
+                    cleaned_dict[k] = v
+
+            return cleaned_dict
+
+        elif isinstance(data, list):
+            pre_cleaned_list = [
+                self._delete_keys_with_empty_value(item, many=many) for item in data
+            ]
+            return [item for item in pre_cleaned_list if item not in self.__empty_values__]
+
+        else:
+            return data