A Django app that allows you to have more control over bulk operations.
Project description
django-bulk-tracker
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:
instance
this is your model instance after it has been updated, or createdchanged_values
is a dict[str, Any] which contains the changed fields only in the case ofpost_update_signal
, in the case ofpost_create_signal
andpost_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:
- Do the above
- You need to inherit your model from
BulkTrackerModel
- 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:
instance
this is your model instance after it has been updated, or createdchanged_values
is a dict[str, Any] which contains the changed fields only in the case ofpost_update_signal
, in the case ofpost_create_signal
andpost_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:
- Do the above
- You need to inherit your model from
BulkTrackerModel
- 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:
instance
this is your model instance after it has been updated, or createdchanged_values
is a dict[str, Any] which contains the changed fields only in the case ofpost_update_signal
, in the case ofpost_create_signal
andpost_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:
- Do the above
- You need to inherit your model from
BulkTrackerModel
- 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
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
Algorithm | Hash digest | |
---|---|---|
SHA256 | c6d088fad3675bf05dc137448a5f3c8a8d99dd25ad740793dca3fdcf68e3af55 |
|
MD5 | 55123af444dbeae355d0c4ec36d1012a |
|
BLAKE2b-256 | af4dbdf62d584f387731a329e3bd8230ab9f446f6facf9190506277e4ef8104f |
File details
Details for the file django_bulk_tracker-0.2.1-py3-none-any.whl
.
File metadata
- Download URL: django_bulk_tracker-0.2.1-py3-none-any.whl
- Upload date:
- Size: 16.6 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.0 CPython/3.12.4
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | c88b7e27ab3b402b1d8b8a36a06396b3b2d8fa0a82fac90287e1c05ec714f0d7 |
|
MD5 | d9321f6598b23fbd609340d79d3f2319 |
|
BLAKE2b-256 | 2125629de017d8a9976d87db4a4c8954e7a86ed8da8472d33f6281f6c55f9d97 |