1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296
|
"""
PolymorphicQuerySet support functions
"""
import copy
from collections import deque
from django.apps import apps
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import FieldDoesNotExist, FieldError
from django.db import models
from django.db.models import Q
from django.db.models.fields.related import ForeignObjectRel, RelatedField
from django.db.utils import DEFAULT_DB_ALIAS
# These functions implement the additional filter- and Q-object functionality.
# They form a kind of small framework for easily adding more
# functionality to filters and Q objects.
# Probably a more general queryset enhancement class could be made out of them.
###################################################################################
# PolymorphicQuerySet support functions
def translate_polymorphic_filter_definitions_in_kwargs(
queryset_model, kwargs, using=DEFAULT_DB_ALIAS
):
"""
Translate the keyword argument list for PolymorphicQuerySet.filter()
Any kwargs with special polymorphic functionality are replaced in the kwargs
dict with their vanilla django equivalents.
For some kwargs a direct replacement is not possible, as a Q object is needed
instead to implement the required functionality. In these cases the kwarg is
deleted from the kwargs dict and a Q object is added to the return list.
Modifies: kwargs dict
Returns: a list of non-keyword-arguments (Q objects) to be added to the filter() query.
"""
additional_args = []
for field_path, val in kwargs.copy().items(): # `copy` so we're not mutating the dict
new_expr = _translate_polymorphic_filter_definition(
queryset_model, field_path, val, using=using
)
if isinstance(new_expr, tuple):
# replace kwargs element
del kwargs[field_path]
kwargs[new_expr[0]] = new_expr[1]
elif isinstance(new_expr, models.Q):
del kwargs[field_path]
additional_args.append(new_expr)
return additional_args
def translate_polymorphic_Q_object(queryset_model, potential_q_object, using=DEFAULT_DB_ALIAS):
def tree_node_correct_field_specs(my_model, node):
"process all children of this Q node"
for i in range(len(node.children)):
child = node.children[i]
if isinstance(child, (tuple, list)):
# this Q object child is a tuple => a kwarg like Q( instance_of=ModelB )
key, val = child
new_expr = _translate_polymorphic_filter_definition(
my_model, key, val, using=using
)
if new_expr:
node.children[i] = new_expr
else:
# this Q object child is another Q object, recursively process this as well
tree_node_correct_field_specs(my_model, child)
if isinstance(potential_q_object, models.Q):
tree_node_correct_field_specs(queryset_model, potential_q_object)
return potential_q_object
def translate_polymorphic_filter_definitions_in_args(queryset_model, args, using=DEFAULT_DB_ALIAS):
"""
Translate the non-keyword argument list for PolymorphicQuerySet.filter()
In the args list, we return all kwargs to Q-objects that contain special
polymorphic functionality with their vanilla django equivalents.
We traverse the Q object tree for this (which is simple).
Returns: modified Q objects
"""
return [
translate_polymorphic_Q_object(queryset_model, copy.deepcopy(q), using=using) for q in args
]
def _translate_polymorphic_filter_definition(
queryset_model, field_path, field_val, using=DEFAULT_DB_ALIAS
):
"""
Translate a keyword argument (field_path=field_val), as used for
PolymorphicQuerySet.filter()-like functions (and Q objects).
A kwarg with special polymorphic functionality is translated into
its vanilla django equivalent, which is returned, either as tuple
(field_path, field_val) or as Q object.
Returns: kwarg tuple or Q object or None (if no change is required)
"""
# handle instance_of expressions or alternatively,
# if this is a normal Django filter expression, return None
if field_path == "instance_of":
return create_instanceof_q(field_val, using=using)
elif field_path == "not_instance_of":
return create_instanceof_q(field_val, not_instance_of=True, using=using)
elif "___" not in field_path:
return None # no change
# filter expression contains '___' (i.e. filter for polymorphic field)
# => get the model class specified in the filter expression
newpath = translate_polymorphic_field_path(queryset_model, field_path)
return (newpath, field_val)
def translate_polymorphic_field_path(queryset_model, field_path):
"""
Translate a field path from a keyword argument, as used for
PolymorphicQuerySet.filter()-like functions (and Q objects).
Supports leading '-' (for order_by args).
E.g.: if queryset_model is ModelA, then "ModelC___field3" is translated
into modela__modelb__modelc__field3.
Returns: translated path (unchanged, if no translation needed)
"""
if not isinstance(field_path, str):
raise ValueError(f"Expected field name as string: {field_path}")
classname, sep, pure_field_path = field_path.partition("___")
if not sep:
return field_path
assert classname, f"PolymorphicModel: {field_path}: bad field specification"
negated = False
if classname[0] == "-":
negated = True
classname = classname.lstrip("-")
if "__" in classname:
# the user has app label prepended to class name via __ => use Django's get_model function
appname, sep, classname = classname.partition("__")
model = apps.get_model(appname, classname)
assert model, f"PolymorphicModel: model {model.__name__} (in app {appname}) not found!"
if not issubclass(model, queryset_model):
e = (
'PolymorphicModel: queryset filter error: "'
+ model.__name__
+ '" is not derived from "'
+ queryset_model.__name__
+ '"'
)
raise AssertionError(e)
else:
# the user has only given us the class name via ___
# => select the model from the sub models of the queryset base model
# Test whether it's actually a regular relation__ _fieldname (the field starting with an _)
# so no tripple ClassName___field was intended.
try:
# This also retreives M2M relations now (including reverse foreign key relations)
field = queryset_model._meta.get_field(classname)
if isinstance(field, (RelatedField, ForeignObjectRel)):
# Can also test whether the field exists in the related object to avoid ambiguity between
# class names and field names, but that never happens when your class names are in CamelCase.
return field_path # No exception raised, field does exist.
except FieldDoesNotExist:
pass
submodels = _get_all_sub_models(queryset_model)
model = submodels.get(classname, None)
assert model, (
f"PolymorphicModel: model {classname} not found (not a subclass of {queryset_model.__name__})!"
)
basepath = _create_base_path(queryset_model, model)
if negated:
newpath = "-"
else:
newpath = ""
newpath += basepath
if basepath:
newpath += "__"
newpath += pure_field_path
return newpath
def _get_all_sub_models(base_model):
"""#Collect all sub-models, this should be optimized (cached)"""
result = {}
queue = deque([base_model])
while queue:
model = queue.popleft()
if issubclass(model, models.Model) and model != models.Model:
# model name is occurring twice in submodel inheritance tree => Error
if model.__name__ in result and model != result[model.__name__]:
name1 = f"{model._meta.app_label}.{model.__name__}"
name2 = (
f"{result[model.__name__]._meta.app_label}.{result[model.__name__].__name__}"
)
raise FieldError(
f"PolymorphicModel: model name alone is ambiguous: {name1} and {name2} match!\n"
f"In this case, please use the syntax: applabel__ModelName___field"
)
result[model.__name__] = model
queue.extend(model.__subclasses__())
return result
def _create_base_path(baseclass, myclass):
# create new field path for expressions, e.g. for baseclass=ModelA, myclass=ModelC
# 'modelb__modelc" is returned
for b in myclass.__bases__:
if b == baseclass:
return _get_query_related_name(myclass)
path = _create_base_path(baseclass, b)
if path:
if b._meta.abstract or b._meta.proxy:
return _get_query_related_name(myclass)
else:
return f"{path}__{_get_query_related_name(myclass)}"
return ""
def _get_query_related_name(myclass):
for f in myclass._meta.local_fields:
if isinstance(f, models.OneToOneField) and f.remote_field.parent_link:
return f.related_query_name()
# Fallback to undetected name,
# this happens on proxy models (e.g. SubclassSelectorProxyModel)
return myclass.__name__.lower()
def create_instanceof_q(modellist, not_instance_of=False, using=DEFAULT_DB_ALIAS):
"""
Helper function for instance_of / not_instance_of
Creates and returns a Q object that filters for the models in modellist,
including all subclasses of these models (as we want to do the same
as pythons isinstance() ).
.
We recursively collect all __subclasses__(), create a Q filter for each,
and or-combine these Q objects. This could be done much more
efficiently however (regarding the resulting sql), should an optimization
be needed.
"""
if not modellist:
return None
if not isinstance(modellist, (list, tuple)):
from .models import PolymorphicModel
if issubclass(modellist, PolymorphicModel):
modellist = [modellist]
else:
raise TypeError(
"PolymorphicModel: instance_of expects a list of (polymorphic) "
"models or a single (polymorphic) model"
)
contenttype_ids = _get_mro_content_type_ids(modellist, using)
q = Q(polymorphic_ctype__in=sorted(contenttype_ids))
if not_instance_of:
q = ~q
return q
def _get_mro_content_type_ids(models, using):
contenttype_ids = set()
for model in models:
ct = ContentType.objects.db_manager(using).get_for_model(model, for_concrete_model=False)
contenttype_ids.add(ct.pk)
subclasses = model.__subclasses__()
if subclasses:
contenttype_ids.update(_get_mro_content_type_ids(subclasses, using))
return contenttype_ids
|