add Validate decorators.

.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/*

README.md

@@ -35,17 +35,18 @@ $ pip install -U db_first
 ### Full example
 
 ```python
-from uuid import UUID
-
 from db_first import BaseCRUD
 from db_first.base_model import ModelMixin
+from db_first.decorators import Validation
 from db_first.mixins import CreateMixin
 from db_first.mixins import DeleteMixin
 from db_first.mixins import ReadMixin
 from db_first.mixins import UpdateMixin
 from marshmallow import fields
 from marshmallow import Schema
 from sqlalchemy import create_engine
+from sqlalchemy import Result
+from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import Mapped
 from sqlalchemy.orm import mapped_column
@@ -64,60 +65,63 @@ class Items(ModelMixin, Base):
 Base.metadata.create_all(engine)
 
 
-class InputSchemaOfCreate(Schema):
-    data = fields.String()
+class IdSchema(Schema):
+    id = fields.UUID()
 
 
-class InputSchemaOfUpdate(InputSchemaOfCreate):
-    id = fields.UUID()
+class SchemaOfCreate(Schema):
+    data = fields.String()
 
 
-class InputSchemaOfRead(Schema):
-    id = fields.UUID()
+class SchemaOfUpdate(IdSchema, SchemaOfCreate):
+    """Update item schema."""
 
 
-class OutputSchema(InputSchemaOfUpdate):
+class OutputSchema(SchemaOfUpdate):
     created_at = fields.DateTime()
 
 
 class ItemController(CreateMixin, ReadMixin, UpdateMixin, DeleteMixin, BaseCRUD):
     class Meta:
         session = session
         model = Items
-        input_schema_of_create = InputSchemaOfCreate
-        input_schema_of_update = InputSchemaOfUpdate
-        output_schema_of_create = OutputSchema
-        input_schema_of_read = InputSchemaOfRead
-        output_schema_of_read = OutputSchema
-        output_schema_of_update = OutputSchema
-        schema_of_paginate = OutputSchema
         sortable = ['created_at']
 
+    @Validation.input(SchemaOfCreate)
+    @Validation.output(OutputSchema, serialize=True)
+    def create(self, **data) -> Result:
+        return super().create_object(**data)
 
-if __name__ == '__main__':
-    item = ItemController()
+    @Validation.input(IdSchema, keys=['id'])
+    @Validation.output(OutputSchema, serialize=True)
+    def read(self, **data) -> Result:
+        return super().read_object(data['id'])
+
+    @Validation.input(SchemaOfUpdate)
+    @Validation.output(OutputSchema, serialize=True)
+    def update(self, **data) -> Result:
+        return super().update_object(**data)
 
-    first_new_item = item.create({'data': 'first'}, deserialize=True)
-    print('Item as object:', first_new_item)
-    second_new_item = item.create({'data': 'second'}, deserialize=True, serialize=True)
-    print('Item as dict:', second_new_item)
+    @Validation.input(IdSchema, keys=['id'])
+    def delete(self, **data) -> None:
+        super().delete_object(**data)
 
-    first_item = item.read({'id': first_new_item.id})
-    print('Item as object:', first_item)
-    first_item = item.read({'id': first_new_item.id})
-    print('Item as dict:', first_item)
 
-    updated_first_item = item.update(data={'id': first_new_item.id, 'data': 'updated_first'})
-    print('Item as object:', updated_first_item)
-    updated_second_item = item.update(
-        data={'id': UUID(second_new_item['id']), 'data': 'updated_second'}, serialize=True
-    )
-    print('Item as dict:', updated_second_item)
+if __name__ == '__main__':
+    item_controller = ItemController()
+
+    new_item = item_controller.create(data='first')
+    print('Item as dict:', new_item)
 
-    items = item.paginate(sort_created_at='desc')
-    print('Items as objects:', items)
-    items = item.paginate(sort_created_at='desc', serialize=True)
-    print('Items as dicts:', items)
+    item = item_controller.read(id=new_item['id'])
+    print('Item as dict:', item)
 
+    updated_item = item_controller.update(id=new_item['id'], data='updated_first')
+    print('Item as dict:', updated_item)
 
+    item_controller.delete(id=new_item['id'])
+    try:
+        item = item_controller.read(id=new_item['id'])
+    except NoResultFound:
+        print('Item deleted:', item)
 ```

examples/full_example.py

@@ -1,14 +1,15 @@
-from uuid import UUID
-
 from db_first import BaseCRUD
 from db_first.base_model import ModelMixin
+from db_first.decorators import Validation
 from db_first.mixins import CreateMixin
 from db_first.mixins import DeleteMixin
 from db_first.mixins import ReadMixin
 from db_first.mixins import UpdateMixin
 from marshmallow import fields
 from marshmallow import Schema
 from sqlalchemy import create_engine
+from sqlalchemy import Result
+from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import Mapped
 from sqlalchemy.orm import mapped_column
@@ -27,57 +28,62 @@ class Items(ModelMixin, Base):
 Base.metadata.create_all(engine)
 
 
-class InputSchemaOfCreate(Schema):
-    data = fields.String()
+class IdSchema(Schema):
+    id = fields.UUID()
 
 
-class InputSchemaOfUpdate(InputSchemaOfCreate):
-    id = fields.UUID()
+class SchemaOfCreate(Schema):
+    data = fields.String()
 
 
-class InputSchemaOfRead(Schema):
-    id = fields.UUID()
+class SchemaOfUpdate(IdSchema, SchemaOfCreate):
+    """Update item schema."""
 
 
-class OutputSchema(InputSchemaOfUpdate):
+class OutputSchema(SchemaOfUpdate):
     created_at = fields.DateTime()
 
 
 class ItemController(CreateMixin, ReadMixin, UpdateMixin, DeleteMixin, BaseCRUD):
     class Meta:
         session = session
         model = Items
-        input_schema_of_create = InputSchemaOfCreate
-        input_schema_of_update = InputSchemaOfUpdate
-        output_schema_of_create = OutputSchema
-        input_schema_of_read = InputSchemaOfRead
-        output_schema_of_read = OutputSchema
-        output_schema_of_update = OutputSchema
-        schema_of_paginate = OutputSchema
         sortable = ['created_at']
 
+    @Validation.input(SchemaOfCreate)
+    @Validation.output(OutputSchema, serialize=True)
+    def create(self, **data) -> Result:
+        return super().create_object(**data)
+
+    @Validation.input(IdSchema, keys=['id'])
+    @Validation.output(OutputSchema, serialize=True)
+    def read(self, **data) -> Result:
+        return super().read_object(data['id'])
+
+    @Validation.input(SchemaOfUpdate)
+    @Validation.output(OutputSchema, serialize=True)
+    def update(self, **data) -> Result:
+        return super().update_object(**data)
+
+    @Validation.input(IdSchema, keys=['id'])
+    def delete(self, **data) -> None:
+        super().delete_object(**data)
+
 
 if __name__ == '__main__':
-    item = ItemController()
-
-    first_new_item = item.create({'data': 'first'}, deserialize=True)
-    print('Item as object:', first_new_item)
-    second_new_item = item.create({'data': 'second'}, deserialize=True, serialize=True)
-    print('Item as dict:', second_new_item)
-
-    first_item = item.read({'id': first_new_item.id})
-    print('Item as object:', first_item)
-    first_item = item.read({'id': first_new_item.id})
-    print('Item as dict:', first_item)
-
-    updated_first_item = item.update(data={'id': first_new_item.id, 'data': 'updated_first'})
-    print('Item as object:', updated_first_item)
-    updated_second_item = item.update(
-        data={'id': UUID(second_new_item['id']), 'data': 'updated_second'}, serialize=True
-    )
-    print('Item as dict:', updated_second_item)
-
-    items = item.paginate(sort_created_at='desc')
-    print('Items as objects:', items)
-    items = item.paginate(sort_created_at='desc', serialize=True)
-    print('Items as dicts:', items)
+    item_controller = ItemController()
+
+    new_item = item_controller.create(data='first')
+    print('Item as dict:', new_item)
+
+    item = item_controller.read(id=new_item['id'])
+    print('Item as dict:', item)
+
+    updated_item = item_controller.update(id=new_item['id'], data='updated_first')
+    print('Item as dict:', updated_item)
+
+    item_controller.delete(id=new_item['id'])
+    try:
+        item = item_controller.read(id=new_item['id'])
+    except NoResultFound:
+        print('Item deleted:', item)

pyproject.toml

@@ -25,7 +25,7 @@ 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 = [
@@ -68,7 +68,7 @@ check-overridden = true
 check-property-returns = true
 check-protected = true
 check-protected-class-methods = true
-disable = ["SIG101"]
+disable = ["SIG101", "SIG501"]
 
 [tool.setuptools.packages.find]
 include = ["db_first*"]

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/base_model.py

@@ -1,13 +1,12 @@
 import uuid
 from datetime import datetime
 from typing import Optional
-from uuid import UUID
 
 from sqlalchemy.orm import Mapped
 from sqlalchemy.orm import mapped_column
 
 
-def make_uuid4() -> UUID:
+def make_uuid4() -> uuid.UUID:
     return uuid.uuid4()
 
 

src/db_first/decorators/__init__.py

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