There are some fine and informative solutions slaready,. but I thoguht I'd share my solution. This solution is a method which can be added to any mode class, to give single instances a rough equivalent to the queryset .update(...) method, with the same arguments syntax. It makes use of the update_fields keyword argument to the model's save method, which enables more efficient behind-the-scenes database updating.
Essentially, it is a wrapper around calling instance.save(...) (whether you have overridden it in your model or not) that behaves the same, argument-wise, as queryset.update(...), and is more effient that calling the "full" save method (for most purposes anyway), and also calls the pre- and post-save signals (but provides an argument for ignoring these like a true queryset update, as well). It also allows passing in an arbitrary number of dictionaries as positional args, which will be automatically converted to keyword args for you.
from django.db.models import ForeignKey, ManyToManyField, OneToOneField, JSONField
from django.contrib.postgres.fields import ArrayField
from django.db.models.manager import Manager
class YourModel(models.Model):
...
def log(self, *args, exception=None, **kwargs):
""" Optionally, usethis method to define how to use your logging setup to log messages related to this
instance specifically. For this SO opost I wuill simply assume 'you've got a logger defined globally,
and this method calls it. But some creative logging could produce highly useful organizational/
informational enhancements. """
(logger.exception if isinstance(exception, Exception) else logger.log)(*args, instance=self, **kwargs)
# Presuming that you've setup your logging to accept
# an instance object, which modifies where it is logged
# to, or something. Feel free to modify this method however
# you see fit.
def update(self, *update_by_dicts, skip_signals=False, use_query=False, refresh_instance=True,
validate_kwargs=False, allow_updating_relations=True, **kwargs):
"""
update instance method
This method enables the calling of instance.update(...) in approximately the same manner
as the update method of a queryset, allowing seamless behavior for updating either a
query or a single instance that's been loaded into memory. It provides options via the
keyword args as described below.
NOTE that providing positional (non-keyword) arguments can be done; if it is, they must each be
a dictionary, which will be unpacked into keywpord arguments as if each key/value pair
had been passed in as a keyword argument to this method.
Args:
skip_signals: If True, then both the pre- and post-save siognals will be ignored after .save is
called at the end of this method's execution (the behavior of a queryset's update method).
You can also pass this in as 'pre' or 'post'; if you do, then the pre_save/ or post_save
signal, respectively, will be skipped, while the other will execute. The default value for
this argument is False, meaning that both pre and post_save are called, like a normal save
method call.
use_query: Normally, this method obviates the need to query "self" (which has already been loaded
fr0om the database into an instance variable, after all) by utilizing the save method, but if
for some reason you wouold prefer not to have this behavior, passing in use_query=True will
cause the method to use a different approach, and will self-query using the ORM, and then call
the typical update method on the resulting one-element queryset. In this case, signals will be
skipped regardless of the value specifyt for the skip_signals argument. However, any positional
argument dicts provided will still be unpacked as passed in as keyword args.
refresh_instance: Only does anything if use_query is True; then, if refresh_instance is True, it will
call refresh_from_db after the ORM update(s), to make sure this instance isn't outdarted. If you're
not goiong to use the instance anymore afterwards, specifying refresh_instance=False saves some time
since it won't re-query it from the database.
validate_kwargs: Normally, all keyword arguments (and keys from positional argument dicts, if any)
will be blindly passed to self.save. Hpwever, if there is any chance that keys supplied may contain
values that do not correspond to existing fields in the model (such as in some system using
polymorphism or other forms of inheritance, where each child model may have some fields unique only to
that model), you can specify validate_fields=True to check all of the fields against their presence in the
instance (using hasattr and discrading if it returns False); specifying a list of argument names instead
will only check those arguments. This adda a nominal amount of overhead time cost to the execution of
Python, so it shoyuld ony be used if it is needed, but it solves a couple of issues related to model
inheritance and/or polymorphism, and protects against dynamic instance updating situations going wrong, too.
allow_updating_relations: If True (which is the default), it enables passing in fields of models accessed through
relations (like ForeignKeys or ManyToManyFields, or their reverse relationships), via standard Djkango query
syntax using double underscores. These updates are done through a normal ORM queryset update, for
efficiency. If False, any argument whose name contains double underscores will not be valid, unless
it is use to reference a key in a JSONField, or an index in an ArrayField.
* PLEASE NOTE that manytomany and reverse foreignkey fields WILL NOT WORK without being validated by providing
validate_kwargs=True or including the related_name relation set manager name in the list provided to validate
kwargs.
kwargs: All other keyword arguments will be interpreted as field names to update, and the values to update them
to. Please note also that any positional argument dicts will be unpacked and literally merged in with
the kwargs dict, with priority in the case of duplicate keys being given to those given in kwargs.
Returns:
A dict of the fields that were not successfully updated with the error message associated with why it failed.
All expected potential exceptions are caught and logged and gracefully handled/logged, to avoid interruption
of your app/program, so the return value can be used to tell you if there were any issues.
"""
if skip_signals is True:
skip_signals = set(['pre_save', 'post_save'])
elif isinstance(skip_signals, str):
skip_signals = set([skip_signals])
elif isinstance(skip_signals, (list, tuple)):
skip_signals = set(skip_signals)
failed_keys = set()
separate_queries = []
self_query_updates = dict()
# Merging all keys and their values from any posotional dictionaries provided, directly into kwargs for ease of
# processing later.
for more_args in update_by_dicts:
for key, val in more_args.items():
if key not in kwargs: # If key was passed explicitly as a kwarg, then we prioritize that and ignore it here
kwargs[key] = val
# If argument validation is desired, we'll do that here, and log any removed after removing those entries from kwargs
if validate_kwargs is True:
# If it is the literal value True, we'll convert it to a list containing the names of every field we requested to update
validate_kwargs = [ key for key in kwargs ]
if validate_kwargs:
# Unless it is False/None, we have at least one field to validate, and will do so now using hasattr, and delete amy
# key/value pairs where the key is not a valid name of an attribute in this instance (or a relation, if applicable).
for field_name in validate_kwargs:
if not (result := check_field_name_validity(self, field_name, value, allow_relations=allow_updating_relations)):
failed_keys.add(field_name)
del kwargs[field_name]
else:
if isinstance(result, dict):
# This is a related query, and the function has returned information about that query
separate_queries.append(result)
del kwargs[field_name] # Deleting from kwargs, since it will be called due to its reference in separate_queries
#elif (result is True) and use_query:
# # Converting any True result to dicts representing ORM queries to use, if use_query argument is True
# separate_queries.append({
# 'type': 'self',
# 'manager': type(self).objects.filter(pk=self.pk),
# 'update_statement': field_name,
# })
# del kwargs[field_name]
if len(kwargs) > 0:
upd_fields = set()
if isinstance(skip_signals, list):
self.__skip_signals = skip_signals
else:
try:
del self.__skip_signals
except:
pass
for field_name, value in kwargs.items():
# Looping through any keys remaining in kwargs in order to modify this instance's fields accordingly, and then call
# self.save(update_fields=[...]) to perform a database UPDATE only on the changed fields for efficiency.
# If one or both save signals are to be skipped, we'll add attributes to the instance; I will lave it to the reader to
# modify the signal receiver(s) to check for the presence of said attributes, and return without doing anything if they're
# present and True.
# If any exceptions are raised, we'll catch, log, and inform the caller in the returned list of failed field names
try:
if use_query:
self_query_updates[field_name] = value # Advantage to using a self query is we don't pre-process, just execute updates as-is
else:
if '__' in field_name:
# traversing the path of indexes/attrbute names for the field, since it has double underwcores. The code below will handle JSONFields, ArrayFields,
# and relations for ForeignKeys and OneToOneFields, in the case that those fields were not validated
path_toks = field_name.split('__')
real_obj = None
if hasattr(self, path_toks[0]) and isinstance(getattr(self, path_toks[0]), models.Model):
result = check_field_name_validity(self, field_name, value, allow_relations=allow_updating_relations)
if not result:
raise ValueError(f"Non-validated kwarg '{field_name}' appears to be a related instance, but there was a problem with the field")
else:
separate_queries.append(result) # If it validates, adding it to separate_queries to avoid code duplication
continue
for attr in path_toks:
real_obj = self if not real_obj else ( else real_obj[last_key])
last_key = int(attr) if attr.isdigit() else attr
else:
upd_fields.add(path_toks[0]) # Add just the root of the field's 'path', as that is the JSONField/etc that we'll update in save()
real_obj[last_key] = value
else:
setattr(self, field_name, value)
upd_fields.add(field_name)
except Exception as e:
self.log("Exception encountered during processing of field '{field_name}'", exception=e)
failed_keys.add(field_name)
if use_query:
# Executing "self-query" by filtering model class for the single instance and using atomic update method of queryset
type(self).objects.filter(pk=self.pk).update(**self_query_updates)
# Then, refreshing the instance from DB so we don't have outdated field values (unless not needed, via arguments)
if refresh_instance:
self.refresh_from_db()
else:
# Finally, calling save with update_fields
self.save(update_fields=upd_fields)
try:
del self.__skip_signals
except:
pass
# Lastly, executing whatever separate queries may have been requested due to related model fields
for qrydef in separate_queries:
try:
qrydef['manager'].update(**qrydef['update_statement'])
except Exception as e:
self.log(f"Exception while processing separarte model query defined by {qrydef}", exception=e)
faled_keys.add(qrydef['key'])
return failed_keys
(Outside of the model class definition)
def check_field_name_validity(instance, field_name, value, allow_relations=True):
if not hasattr(instance, field_name):
if '__' in field_name:
path_tokens = field_name.split('__')
arg_valid = False
try:
if hasattr(instance, path_tokens[0]):
# Usage of isinstance allows subclasses of these field types to be recognized
obj = getattr(instance, path_tokens[0])
if isinstance(instance._meta.get_field(path_tokens[0]), JSONField):
if len(path_tokens) > 1:
for nextkey in path_tokens[1:]:
obj = obj[nextkey]
# If we've made it to the end of the path of keys, then this argument is valid
return True
elif isinstance(instance._meta.get_field(path_tokens[0]), ArrayField):
value = obj
for index in [ (int(x) if x.isdigit() else x) for x in path_tokens[1:] ]:
# We converted each path "token" from the split on double underscore into
# an integer if it is a str representation of one, else left it as a str;
# this allows nested ArrayFields or ArrayFields made up of DictFields.
value = value[index]
# If we've made it to the end of the path of keys, then this argument is valid
return True
elif isinstance(instance._meta.get_field(path_tokens[0]), Manager):
# The fact that its class is Manager means it is a reverse relation, so we'll
# need to check the validity of the rest of it by seeing if a query on it results
# in an exception. We'll use __isnull since it should work for any valid field
if not allow_relations:
raise TypeError(f"Field '{path_tokens[0]}' of field name '{field_name}' is a related set manager, but allow_relations is False")
search_path = "__".join(path_tokens[1:])
try:
if not obj.filter(**{qry_path + '__isnull'): False).exists():
raise ValueError(f"Relation field argument '{field_name}' does not exist or the query results in no matches")
except Exception as e:
raise e
else:
return {
'key': field_name,
'type': 'collection',
'manager': obj,
'update_statement': {"__".join(path_tokens[1:]): value},
}
else:
# We'll assume it is a OneToOne/ForeignKey field and if it's not it will error, which tells us it's invalid.
# 'obj' should contain the followed reference to the related instance already, if so.
# We'll recursively call this function on the reated object and return the result. Recursion is a beautiful thing!
# If you don't know, now you know.
if not allow_relations:
raise TypeError(f"Field '{path_tokens[0]}' of field name '{field_name}' is a related model instance, but allow_relations is False")
arg_valid = check_field_name_validity(obj, "__".join(path_tokens)[1:], value, allow_relations=True)
if arg_valid:
return {
'key': field_name,
'type': 'instance',
'manager': type(obj).objects.filter(pk=obj.pk),
'update_statement': {"__".join(path_tokens[1:]): value},
}
else:
raise ValueError(f"Field '{field_name}' cannot be found in the instance nor any related managers or instances")
except Exception as e:
# If any exception is caught at all, it means this is not a valid argument, and we'll log the exception and remove it from kwargs
instance.log(f"Field '{field_name}' is invalid. Reason = {e}", exception=e)
return False
*NOTE: this 'is similar to what I use in my web platform I've been developing for the past couple of years, but I made changes that
I havcen't tested and am prone to typos. If you use this and find issues, please let me know and I'll fix them in this post.
