feat: add on_update to the field

README.md

@@ -655,6 +655,7 @@ The following keyword arguments are supported on all field types.
 * `choices: typing.Sequence`
 * `name: str`
 * `pydantic_only: bool`
+* `on_update: Any/callable`
 
 All fields are required unless one of the following is set:
 

docs/fields/common-parameters.md

@@ -216,6 +216,40 @@ class OverwriteTest(ormar.Model):
 
 `choices`: `Sequence` = `[]` 
 
+## on_update
+
+when the object update or bulk_update, if you don't update the field which has the on_update option,
+its value will be changed from `on_update definition`
+
+```python
+
+class ToDo(ormar.Model):
+    class Meta:
+        tablename = "todo"
+        metadata = metadata
+        database = database
+
+    id: int = ormar.Integer(primary_key=True)
+    name: str = ormar.String(
+        max_length=255,
+        on_update=lambda: "hello",
+    )
+    is_dirty: bool = ormar.Boolean(default=False, on_update=True)
+    updated_at: datetime = ormar.DateTime(
+        default=datetime.now, server_default=func.now(), on_update=datetime.now
+    )
+
+await ToDo.objects.create(name="test")
+todo = await ToDo.objects.get(id=1)
+await todo.update()
+
+assert todo.is_dirty
+assert todo.name == "hello"
+
+```
+
+
+
 A set of choices allowed to be used for given field.
 
 Used for data validation on pydantic side.

ormar/fields/base.py

@@ -1,5 +1,5 @@
 import warnings
-from typing import Any, Dict, List, Optional, TYPE_CHECKING, Type, Union
+from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING, Type, Union
 
 import sqlalchemy
 from pydantic import Json, typing
@@ -47,6 +47,9 @@ def __init__(self, **kwargs: Any) -> None:
         self.index: bool = kwargs.pop("index", False)
         self.unique: bool = kwargs.pop("unique", False)
         self.pydantic_only: bool = kwargs.pop("pydantic_only", False)
+        self.on_update: Union[Callable[..., Any], Any, None] = kwargs.pop(
+            "on_update", None
+        )
         if self.pydantic_only:
             warnings.warn(
                 "Parameter `pydantic_only` is deprecated and will "
@@ -217,6 +220,23 @@ def has_default(self, use_server: bool = True) -> bool:
             self.server_default is not None and use_server
         )
 
+    def has_onupdate(self) -> bool:
+        """
+        Checks if the field has onupdate value set.
+        :return: result of the check if onupdate value is set
+        rtype: bool
+        """
+        return self.on_update is not None
+
+    def get_onupdate(self) -> Union[None, Any]:
+        """
+        Get onupdate value if set
+
+        :return: result of the onupdate
+        rtype: Any
+        """
+        return self.on_update() if callable(self.on_update) else self.on_update
+
     def is_auto_primary_key(self) -> bool:
         """
         Checks if field is first a primary key and if it,

ormar/models/metaclass.py

@@ -111,6 +111,7 @@ def add_cached_properties(new_model: Type["Model"]) -> None:
     new_model._pydantic_fields = {name for name in new_model.__fields__}
     new_model._json_fields = set()
     new_model._bytes_fields = set()
+    new_model._onupdate_fields = set()
 
 
 def add_property_fields(new_model: Type["Model"], attrs: Dict) -> None:  # noqa: CCR001

ormar/models/mixins/save_mixin.py

@@ -22,6 +22,7 @@
 from ormar.models.mixins.relation_mixin import RelationMixin
 
 if TYPE_CHECKING:  # pragma: no cover
+    from ormar.models import T
     from ormar import ForeignKeyField, Model
 
 
@@ -35,6 +36,7 @@ class SavePrepareMixin(RelationMixin, AliasMixin):
         _skip_ellipsis: Callable
         _json_fields: Set[str]
         _bytes_fields: Set[str]
+        _onupdate_fields: Set[str]
         __fields__: Dict[str, pydantic.fields.ModelField]
 
     @classmethod
@@ -242,6 +244,27 @@ def populate_default_values(cls, new_kwargs: Dict) -> Dict:
                 new_kwargs.pop(field_name, None)
         return new_kwargs
 
+    @classmethod
+    def populate_onupdate_value(cls, new_kwargs: Dict, obj: "T" = None) -> Dict:
+        """
+        Populate value which from onupdate options in field
+
+        :param new_kwargs: dictionary of model that is about to be saved
+        :type new_kwargs: Dict
+        :param obj: ormar models
+        :type obj: Model
+        :return: dictionary of model that is about to be saved
+        :rtype: Dict
+        """
+        for field_name in cls.get_fields_with_onupdate():
+            field = cls.Meta.model_fields[field_name]
+            if field_name not in new_kwargs:
+                new_kwargs[field_name] = field.get_onupdate()
+            if obj:
+                if field_name not in obj.__setattr_fields__:
+                    new_kwargs[field_name] = field.get_onupdate()
+        return new_kwargs
+
     @classmethod
     def validate_choices(cls, new_kwargs: Dict) -> Dict:
         """
@@ -406,3 +429,13 @@ def _get_field_values(self, name: str) -> List:
         if not isinstance(values, list):
             values = [values]
         return values
+
+    @classmethod
+    def get_fields_with_onupdate(cls) -> Set[str]:
+        if not cls._onupdate_fields:
+            cls._onupdate_fields = {
+                field_name
+                for field_name, field in cls.Meta.model_fields.items()
+                if field.has_onupdate() and not field.pydantic_only
+            }
+        return cls._onupdate_fields

ormar/models/model.py

@@ -229,6 +229,7 @@ async def update(self: T, _columns: List[str] = None, **kwargs: Any) -> T:
         :rtype: Model
         """
         if kwargs:
+            kwargs = self.populate_onupdate_value(kwargs)
             self.update_from_dict(kwargs)
 
         if not self.pk:
@@ -242,7 +243,19 @@ async def update(self: T, _columns: List[str] = None, **kwargs: Any) -> T:
         self_fields = self._extract_model_db_fields()
         self_fields.pop(self.get_column_name_from_alias(self.Meta.pkname))
         if _columns:
-            self_fields = {k: v for k, v in self_fields.items() if k in _columns}
+            onupdate_fields = self.get_fields_with_onupdate()
+            self_fields = {
+                k: v
+                for k, v in self_fields.items()
+                if k in _columns or k in onupdate_fields
+            }
+        if not kwargs and not _columns:
+            for field_name in self.get_fields_with_onupdate():
+                if field_name not in self.__setattr_fields__:
+                    field = self.Meta.model_fields[field_name]
+                    onupdate_field_value = {field_name: field.get_onupdate()}
+                    self_fields.update(onupdate_field_value)
+                    self.update_from_dict(onupdate_field_value)
         self_fields = self.translate_columns_to_aliases(self_fields)
         expr = self.Meta.table.update().values(**self_fields)
         expr = expr.where(self.pk_column == getattr(self, self.Meta.pkname))

ormar/models/newbasemodel.py

@@ -74,6 +74,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
         "_pk_column",
         "__pk_only__",
         "__cached_hash__",
+        "__setattr_fields__",
     )
 
     if TYPE_CHECKING:  # pragma no cover
@@ -87,6 +88,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
         __database__: databases.Database
         __relation_map__: Optional[List[str]]
         __cached_hash__: Optional[int]
+        __setattr_fields__: Set
         _orm_relationship_manager: AliasManager
         _orm: RelationsManager
         _orm_id: int
@@ -99,6 +101,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
         _quick_access_fields: Set
         _json_fields: Set
         _bytes_fields: Set
+        _onupdate_fields: Set
         Meta: ModelMeta
 
     # noinspection PyMissingConstructor
@@ -168,6 +171,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:  # type: ignore
         if hasattr(self, "_init_private_attributes"):
             # introduced in pydantic 1.7
             self._init_private_attributes()
+        object.__setattr__(self, "__setattr_fields__", set())
 
     def __setattr__(self, name: str, value: Any) -> None:  # noqa CCR001
         """
@@ -195,6 +199,7 @@ def __setattr__(self, name: str, value: Any) -> None:  # noqa CCR001
 
             if prev_hash != new_hash:
                 self._update_relation_cache(prev_hash, new_hash)
+        self.__setattr_fields__.add(name)
 
     def __getattr__(self, item: str) -> Any:
         """

ormar/queryset/join.py

@@ -92,7 +92,14 @@ def to_table(self) -> sqlalchemy.Table:
         """
         return self.next_model.Meta.table
 
-    def _on_clause(self, previous_alias: str, from_table_name:str, from_column_name: str, to_table_name: str, to_column_name: str) -> text:
+    def _on_clause(
+        self,
+        previous_alias: str,
+        from_table_name: str,
+        from_column_name: str,
+        to_table_name: str,
+        to_column_name: str,
+    ) -> text:
         """
         Receives aliases and names of both ends of the join and combines them
         into one text clause used in joins.
@@ -112,11 +119,15 @@ def _on_clause(self, previous_alias: str, from_table_name:str, from_column_name:
         """
         dialect = self.main_model.Meta.database._backend._dialect
         quoter = dialect.identifier_preparer.quote
-        left_part = f"{quoter(f'{self.next_alias}_{to_table_name}')}.{quoter(to_column_name)}"
+        left_part = (
+            f"{quoter(f'{self.next_alias}_{to_table_name}')}.{quoter(to_column_name)}"
+        )
         if not previous_alias:
             right_part = f"{quoter(from_table_name)}.{quoter(from_column_name)}"
         else:
-            right_part = f"{quoter(f'{previous_alias}_{from_table_name}')}.{from_column_name}"
+            right_part = (
+                f"{quoter(f'{previous_alias}_{from_table_name}')}.{from_column_name}"
+            )
 
         return text(f"{left_part}={right_part}")
 

ormar/queryset/queryset.py

@@ -796,6 +796,7 @@ async def update(self, each: bool = False, **kwargs: Any) -> int:
             self.model.extract_related_names()
         )
         updates = {k: v for k, v in kwargs.items() if k in self_fields}
+        updates = self.model.populate_onupdate_value(updates)
         updates = self.model.validate_choices(updates)
         updates = self.model.translate_columns_to_aliases(updates)
 
@@ -1190,7 +1191,12 @@ async def bulk_update(  # noqa:  CCR001
         if pk_name not in columns:
             columns.append(pk_name)
 
-        columns = [self.model.get_column_alias(k) for k in columns]
+        columns = {self.model.get_column_alias(k) for k in columns}
+        on_update_fields = {
+            self.model.get_column_alias(k)
+            for k in cast(Type["Model"], self.model_cls).get_fields_with_onupdate()
+        }
+        columns |= on_update_fields
 
         for obj in objects:
             new_kwargs = obj.dict()
@@ -1199,6 +1205,7 @@ async def bulk_update(  # noqa:  CCR001
                     "You cannot update unsaved objects. "
                     f"{self.model.__name__} has to have {pk_name} filled."
                 )
+            new_kwargs = obj.populate_onupdate_value(new_kwargs, obj)
             new_kwargs = obj.prepare_model_to_update(new_kwargs)
             ready_objects.append(
                 {"new_" + k: v for k, v in new_kwargs.items() if k in columns}

tests/test_model_definition/test_field_quoting.py

@@ -41,8 +41,12 @@ class Meta:
     id: int = ormar.Integer(primary_key=True)
     name: str = ormar.String(max_length=100)
     gpa: float = ormar.Float()
-    schoolclass: Optional[SchoolClass] = ormar.ForeignKey(SchoolClass, related_name="students")
-    category: Optional[Category] = ormar.ForeignKey(Category, nullable=True, related_name="students")
+    schoolclass: Optional[SchoolClass] = ormar.ForeignKey(
+        SchoolClass, related_name="students"
+    )
+    category: Optional[Category] = ormar.ForeignKey(
+        Category, nullable=True, related_name="students"
+    )
 
 
 @pytest.fixture(autouse=True, scope="module")
@@ -59,9 +63,15 @@ async def create_data():
     class2 = await SchoolClass.objects.create(name="Logic")
     category = await Category.objects.create(name="Foreign")
     category2 = await Category.objects.create(name="Domestic")
-    await Student.objects.create(name="Jane", category=category, schoolclass=class1, gpa=3.2)
-    await Student.objects.create(name="Judy", category=category2, schoolclass=class1, gpa=2.6)
-    await Student.objects.create(name="Jack", category=category2, schoolclass=class2, gpa=3.8)
+    await Student.objects.create(
+        name="Jane", category=category, schoolclass=class1, gpa=3.2
+    )
+    await Student.objects.create(
+        name="Judy", category=category2, schoolclass=class1, gpa=2.6
+    )
+    await Student.objects.create(
+        name="Jack", category=category2, schoolclass=class2, gpa=3.8
+    )
 
 
 @pytest.mark.asyncio
@@ -70,10 +80,14 @@ async def test_quotes_left_join():
         async with database.transaction(force_rollback=True):
             await create_data()
             students = await Student.objects.filter(
-                (Student.schoolclass.name == "Math") | (Student.category.name == "Foreign")
+                (Student.schoolclass.name == "Math")
+                | (Student.category.name == "Foreign")
             ).all()
             for student in students:
-                assert student.schoolclass.name == "Math" or student.category.name == "Foreign"
+                assert (
+                    student.schoolclass.name == "Math"
+                    or student.category.name == "Foreign"
+                )
 
 
 @pytest.mark.asyncio
@@ -92,8 +106,9 @@ async def test_quotes_deep_join():
     async with database:
         async with database.transaction(force_rollback=True):
             await create_data()
-            schoolclasses = await SchoolClass.objects.filter(students__category__name="Domestic").all()
+            schoolclasses = await SchoolClass.objects.filter(
+                students__category__name="Domestic"
+            ).all()
             for schoolclass in schoolclasses:
                 for student in schoolclass.students:
                     assert student.category.name == "Domestic"
-

tests/test_model_methods/test_populate_default_values.py

@@ -1,5 +1,4 @@
 import databases
-import pytest
 import sqlalchemy
 from sqlalchemy import text