Skip to content

Commit

Permalink
feat: model signal helpers (#19)
Browse files Browse the repository at this point in the history
* UpdateFields type

* feat: model signal helpers

* fix path ignore

* remove

* ignore . files

* escape .

* ignore . 2

* test

* test2

* final

* models helpers
  • Loading branch information
SKairinos authored Nov 14, 2023
1 parent 7a1b07a commit 476c777
Show file tree
Hide file tree
Showing 5 changed files with 126 additions and 1 deletion.
3 changes: 2 additions & 1 deletion .github/workflows/main.yml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@ on:
paths-ignore:
- "codeforlife/version.py"
- "**/*.md"
- "**/.*"
- ".vscode/**"
- ".*"
workflow_dispatch:

env:
Expand Down
21 changes: 21 additions & 0 deletions codeforlife/models/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
"""Helpers for module "django.db.models".
https://docs.djangoproject.com/en/3.2/ref/models/
"""

import typing as t

from django.db.models import Model as _Model


class Model(_Model):
"""A base class for all Django models.
Args:
_Model (django.db.models.Model): Django's model class.
"""

id: int
pk: int


AnyModel = t.TypeVar("AnyModel", bound=Model)
16 changes: 16 additions & 0 deletions codeforlife/models/signals/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
"""Helpers for module "django.db.models.signals".
https://docs.djangoproject.com/en/3.2/ref/signals/#module-django.db.models.signals
"""

import typing as t

UpdateFields = t.Optional[t.FrozenSet[str]]


def _has_update_fields(actual: UpdateFields, expected: UpdateFields):
if expected is None:
return actual is None
if actual is None:
return False

return all(update_field in actual for update_field in expected)
21 changes: 21 additions & 0 deletions codeforlife/models/signals/post_save.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
"""Helpers for module "django.db.models.signals.post_save".
https://docs.djangoproject.com/en/3.2/ref/signals/#post-save
"""

from . import UpdateFields, _has_update_fields


def has_update_fields(actual: UpdateFields, expected: UpdateFields):
"""Check if the expected fields were updated.
Args:
actual: The fields that were updated.
expected: A subset of the fields that were expected to be updated. If no
fields were expected to be updated, set to None.
Returns:
If the fields that were expected to be updated are a subset of the
fields that were updated.
"""

return _has_update_fields(actual, expected)
66 changes: 66 additions & 0 deletions codeforlife/models/signals/pre_save.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
"""Helpers for module "django.db.models.signals.pre_save".
https://docs.djangoproject.com/en/3.2/ref/signals/#pre-save
"""

import typing as t

from .. import AnyModel
from . import UpdateFields, _has_update_fields


def was_created(instance: AnyModel):
"""Check if the instance was created.
Args:
instance: The current instance.
Returns:
If the instance was created.
"""

return instance.pk is not None


def has_update_fields(actual: UpdateFields, expected: UpdateFields):
"""Check if the expected fields are going to be updated.
Args:
actual: The fields that are going to be updated.
expected: A subset of the fields that are expected to be updated. If no
fields are expected to be updated, set to None.
Returns:
If the fields that are expected to be updated are a subset of the
fields that are going to be updated.
"""

return _has_update_fields(actual, expected)


def has_previous_values(
instance: AnyModel,
predicates: t.Dict[str, t.Callable[[t.Any, t.Any], bool]],
):
"""Check if the previous values are as expected.
Args:
instance: The current instance.
predicates: A predicate for each field. It accepts the arguments
(previous_value, value) and returns True if the values are as expected.
Raises:
ValueError: If arg 'instance' has not been created yet.
Returns:
If all the previous values are as expected.
"""

if not was_created(instance):
raise ValueError("Arg 'instance' has not been created yet.")

previous_instance = instance.__class__.objects.get(pk=instance.pk)

return all(
predicate(previous_instance[field], instance[field])
for field, predicate in predicates.items()
)

0 comments on commit 476c777

Please sign in to comment.