File: query_translate.py

package info (click to toggle)
django-polymorphic 4.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 892 kB
  • sloc: python: 6,784; javascript: 263; makefile: 137
file content (296 lines) | stat: -rw-r--r-- 11,148 bytes parent folder | download
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