Skip to main content

A Django app that allows you to have more control over bulk operations.

Project description

django-bulk-tracker

Build Status Build Status codecov Supported Python versions Supported Django versions

Documentation

https://django-bulk-tracker.readthedocs.io/en/latest/

Run tests

pip install -r requirements_dev.txt
pytest

Here's the provided text translated from reStructuredText (.rst) format to Markdown (.md):

Usage

django-bulk-tracker will emit a signal whenever you update, create, or delete a record in the database.

django-bulk-tracker supports bulk operations:

  • queryset.update()
  • queryset.bulk_update()
  • queryset.bulk_create()
  • queryset.delete()

and single operations:

  • create()
  • save() # update and create
  • delete()

All you need to do is define your Model and inherit from:

from bulk_tracker.models import BulkTrackerModel

class Author(BulkTrackerModel):
    first_name = models.CharField(max_length=200)
    last_name = models.CharField(max_length=200)

OR if you have a custom queryset inherit from or Don't want to support single-operation:

from bulk_tracker.managers import BulkTrackerQuerySet

class MyModelQuerySet(BulkTrackerQuerySet):
    def do_something_custom(self):
        pass

Now you can listen to the signals post_update_signal, post_create_signal, post_delete_signal:

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    do_stuff()

Hint: All signals have the same signature for consistency and also in case you want to assign one function to listen to multiple signals.

ModifiedObject is a very simple object, it contains 2 attributes:

  1. instance this is your model instance after it has been updated, or created
  2. changed_values is a dict[str, Any] which contains the changed fields only in the case of post_update_signal, in the case of post_create_signal and post_delete_signal, changed_values will be an empty dict {}.

Optionally you can pass tracking_info_ to your functions, as in:

from bulk_tracker.helper_objects import TrackingInfo

def a_function_that_updates_records():
    user = self.request.user
    MyModel.objects.filter(name='john').update(
        name='jack',
        tracking_info_=TrackingInfo(user=user, comment="Updated from a function", kwargs={'app-build':'1.1.8'}, is_robust=True),
        )

Hint: tracking_info_ has a trailing underscore to avoid collision with your actual fields. You can use TrackingInfo to implement any kind of behavior like logging in your signal handlers and you need to capture more info about the operation that is happening.

For single operations as well

  • create()
  • save() # update and create
  • delete()

To support, we rely on the amazing library django-model-utils to track the model instances:

  1. Do the above
  2. You need to inherit your model from BulkTrackerModel
  3. Add tracker = FieldTracker() to your model

As in:

from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

class MyModel(BulkTrackerModel):
    objects = MyModelQuerySet.as_manager() # MyModelManager() if you have
    tracker = FieldTracker()

Robust Send

robust_send if you have multiple receivers for the same signal, and you want to make sure that all of them are executed, even if one of them raises an exception. You can add TrackingInfo(is_robust=True) in your operation. You can read more about robust_send in the official documentation.

As in:

MyModel.objects.filter(name='john').update(
    name='jack',
    tracking_info_=TrackingInfo(is_robust=True),
)

Complete Example

# models.py
from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

from myapp.managers import MyModelManager

class MyModel(BulkTrackerModel):
    first_field = models.CharField()
    second_field = models.CharField()

    objects = MyModelManager()
    tracker = FieldTracker()
# managers.py
from bulk_tracker.managers import BulkTrackerQuerySet # optional

class MyModelQuerySet(BulkTrackerQuerySet):
    pass

class MyModelManager(BulkTrackerManager.from_queryset(MyModelQuerySet)): # optional
    pass
# signal_handlers.py
from bulk_tracker.signals import post_update_signal
from bulk_tracker.helper_objects import ModifiedObject, TrackingInfo

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    user = tracking_info_.user if tracking_info_ else None
    for modified_object in modified_objects:
        if 'name' in modified_object.changed_values:
            log(f"field 'name' has changed by {user or ''}")
            notify_user()

Usage

django-bulk-tracker will emit a signal whenever you update, create, or delete a record in the database.

django-bulk-tracker supports bulk operations:

  • queryset.update()
  • queryset.bulk_update()
  • queryset.bulk_create()
  • queryset.delete()

and single operations:

  • create()
  • save() # update and create
  • delete()

All you need to do is define your Model and inherit from:

from bulk_tracker.models import BulkTrackerModel

class Author(BulkTrackerModel):
    first_name = models.CharField(max_length=200)
    last_name = models.CharField(max_length=200)

OR if you have a custom queryset inherit from or Don't want to support single-operation:

from bulk_tracker.managers import BulkTrackerQuerySet

class MyModelQuerySet(BulkTrackerQuerySet):
    def do_something_custom(self):
        pass

Now you can listen to the signals post_update_signal, post_create_signal, post_delete_signal:

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    do_stuff()

Hint: All signals have the same signature for consistency and also in case you want to assign one function to listen to multiple signals.

ModifiedObject is a very simple object, it contains 2 attributes:

  1. instance this is your model instance after it has been updated, or created
  2. changed_values is a dict[str, Any] which contains the changed fields only in the case of post_update_signal, in the case of post_create_signal and post_delete_signal, changed_values will be an empty dict {}.

Optionally you can pass tracking_info_ to your functions, as in:

from bulk_tracker.helper_objects import TrackingInfo

def a_function_that_updates_records():
    user = self.request.user
    MyModel.objects.filter(name='john').update(
        name='jack',
        tracking_info_=TrackingInfo(user=user, comment="Updated from a function", kwargs={'app-build':'1.1.8'}, is_robust=True),
        )

Hint: tracking_info_ has a trailing underscore to avoid collision with your actual fields. You can use TrackingInfo to implement any kind of behavior like logging in your signal handlers and you need to capture more info about the operation that is happening.

For single operations as well

  • create()
  • save() # update and create
  • delete()

To support, we rely on the amazing library django-model-utils to track the model instances:

  1. Do the above
  2. You need to inherit your model from BulkTrackerModel
  3. Add tracker = FieldTracker() to your model

As in:

from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

class MyModel(BulkTrackerModel):
    objects = MyModelQuerySet.as_manager() # MyModelManager() if you have
    tracker = FieldTracker()

Robust Send

robust_send if you have multiple receivers for the same signal, and you want to make sure that all of them are executed, even if one of them raises an exception. You can add TrackingInfo(is_robust=True) in your operation. You can read more about robust_send in the official documentation.

As in:

MyModel.objects.filter(name='john').update(
    name='jack',
    tracking_info_=TrackingInfo(is_robust=True),
)

Complete Example

# models.py
from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

from myapp.managers import MyModelManager

class MyModel(BulkTrackerModel):
    first_field = models.CharField()
    second_field = models.CharField()

    objects = MyModelManager()
    tracker = FieldTracker()
# managers.py
from bulk_tracker.managers import BulkTrackerQuerySet # optional

class MyModelQuerySet(BulkTrackerQuerySet):
    pass

class MyModelManager(BulkTrackerManager.from_queryset(MyModelQuerySet)): # optional
    pass
# signal_handlers.py
from bulk_tracker.signals import post_update_signal
from bulk_tracker.helper_objects import ModifiedObject, TrackingInfo

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    user = tracking_info_.user if tracking_info_ else None
    for modified_object in modified_objects:
        if 'name' in modified_object.changed_values:
            log(f"field 'name' has changed by {user or ''}")
            notify_user()

Usage

django-bulk-tracker will emit a signal whenever you update, create, or delete a record in the database.

django-bulk-tracker supports bulk operations:

  • queryset.update()
  • queryset.bulk_update()
  • queryset.bulk_create()
  • queryset.delete()

and single operations:

  • create()
  • save() # update and create
  • delete()

All you need to do is define your Model and inherit from:

from bulk_tracker.models import BulkTrackerModel

class Author(BulkTrackerModel):
    first_name = models.CharField(max_length=200)
    last_name = models.CharField(max_length=200)

OR if you have a custom queryset inherit from or Don't want to support single-operation:

from bulk_tracker.managers import BulkTrackerQuerySet

class MyModelQuerySet(BulkTrackerQuerySet):
    def do_something_custom(self):
        pass

Now you can listen to the signals post_update_signal, post_create_signal, post_delete_signal:

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    do_stuff()

Hint: All signals have the same signature for consistency and also in case you want to assign one function to listen to multiple signals.

ModifiedObject is a very simple object, it contains 2 attributes:

  1. instance this is your model instance after it has been updated, or created
  2. changed_values is a dict[str, Any] which contains the changed fields only in the case of post_update_signal, in the case of post_create_signal and post_delete_signal, changed_values will be an empty dict {}.

Optionally you can pass tracking_info_ to your functions, as in:

from bulk_tracker.helper_objects import TrackingInfo

def a_function_that_updates_records():
    user = self.request.user
    MyModel.objects.filter(name='john').update(
        name='jack',
        tracking_info_=TrackingInfo(user=user, comment="Updated from a function", kwargs={'app-build':'1.1.8'}, is_robust=True),
        )

Hint: tracking_info_ has a trailing underscore to avoid collision with your actual fields. You can use TrackingInfo to implement any kind of behavior like logging in your signal handlers and you need to capture more info about the operation that is happening.

For single operations as well

  • create()
  • save() # update and create
  • delete()

To support, we rely on the amazing library django-model-utils to track the model instances:

  1. Do the above
  2. You need to inherit your model from BulkTrackerModel
  3. Add tracker = FieldTracker() to your model

As in:

from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

class MyModel(BulkTrackerModel):
    objects = MyModelQuerySet.as_manager() # MyModelManager() if you have
    tracker = FieldTracker()

Robust Send

robust_send if you have multiple receivers for the same signal, and you want to make sure that all of them are executed, even if one of them raises an exception. You can add TrackingInfo(is_robust=True) in your operation. You can read more about robust_send in the official documentation.

As in:

MyModel.objects.filter(name='john').update(
    name='jack',
    tracking_info_=TrackingInfo(is_robust=True),
)

Complete Example

# models.py
from bulk_tracker.models import BulkTrackerModel
from model_utils import FieldTracker

from myapp.managers import MyModelManager

class MyModel(BulkTrackerModel):
    first_field = models.CharField()
    second_field = models.CharField()

    objects = MyModelManager()
    tracker = FieldTracker()
# managers.py
from bulk_tracker.managers import BulkTrackerQuerySet # optional

class MyModelQuerySet(BulkTrackerQuerySet):
    pass

class MyModelManager(BulkTrackerManager.from_queryset(MyModelQuerySet)): # optional
    pass
# signal_handlers.py
from bulk_tracker.signals import post_update_signal
from bulk_tracker.helper_objects import ModifiedObject, TrackingInfo

@receiver(post_update_signal, sender=MyModel)
def i_am_a_receiver_function(
    sender,
    objects: list[ModifiedObject[MyModel]],
    tracking_info_: TrackingInfo | None = None,
    **kwargs,
):
    user = tracking_info_.user if tracking_info_ else None
    for modified_object in modified_objects:
        if 'name' in modified_object.changed_values:
            log(f"field 'name' has changed by {user or ''}")
            notify_user()

Contribute

If you have great ideas for django-bulk-tracker, or if you like to improve something, feel free to fork this repository and/or create a pull request. I'm open for suggestions. If you like to discuss something with me (about django-bulk-tracker), please open an issue.

GitHub repository: https://github.com/hassaanalansary/django-bulk-tracker

issue tracker: https://github.com/hassaanalansary/django-bulk-tracker/issues

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

django_bulk_tracker-0.2.1.tar.gz (19.8 kB view details)

Uploaded Source

Built Distribution

django_bulk_tracker-0.2.1-py3-none-any.whl (16.6 kB view details)

Uploaded Python 3

File details

Details for the file django_bulk_tracker-0.2.1.tar.gz.

File metadata

  • Download URL: django_bulk_tracker-0.2.1.tar.gz
  • Upload date:
  • Size: 19.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.0 CPython/3.12.4

File hashes

Hashes for django_bulk_tracker-0.2.1.tar.gz
Algorithm Hash digest
SHA256 c6d088fad3675bf05dc137448a5f3c8a8d99dd25ad740793dca3fdcf68e3af55
MD5 55123af444dbeae355d0c4ec36d1012a
BLAKE2b-256 af4dbdf62d584f387731a329e3bd8230ab9f446f6facf9190506277e4ef8104f

See more details on using hashes here.

File details

Details for the file django_bulk_tracker-0.2.1-py3-none-any.whl.

File metadata

File hashes

Hashes for django_bulk_tracker-0.2.1-py3-none-any.whl
Algorithm Hash digest
SHA256 c88b7e27ab3b402b1d8b8a36a06396b3b2d8fa0a82fac90287e1c05ec714f0d7
MD5 d9321f6598b23fbd609340d79d3f2319
BLAKE2b-256 2125629de017d8a9976d87db4a4c8954e7a86ed8da8472d33f6281f6c55f9d97

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page