Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/db/models/sql/datastructures.py: 74%
92 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""
2Useful auxiliary data structures for query construction. Not useful outside
3the SQL domain.
4"""
5from django.db.models.sql.constants import INNER, LOUTER
8class MultiJoin(Exception):
9 """
10 Used by join construction code to indicate the point at which a
11 multi-valued join was attempted (if the caller wants to treat that
12 exceptionally).
13 """
15 def __init__(self, names_pos, path_with_names):
16 self.level = names_pos
17 # The path travelled, this includes the path to the multijoin.
18 self.names_with_path = path_with_names
21class Empty:
22 pass
25class Join:
26 """
27 Used by sql.Query and sql.SQLCompiler to generate JOIN clauses into the
28 FROM entry. For example, the SQL generated could be
29 LEFT OUTER JOIN "sometable" T1
30 ON ("othertable"."sometable_id" = "sometable"."id")
32 This class is primarily used in Query.alias_map. All entries in alias_map
33 must be Join compatible by providing the following attributes and methods:
34 - table_name (string)
35 - table_alias (possible alias for the table, can be None)
36 - join_type (can be None for those entries that aren't joined from
37 anything)
38 - parent_alias (which table is this join's parent, can be None similarly
39 to join_type)
40 - as_sql()
41 - relabeled_clone()
42 """
44 def __init__(
45 self,
46 table_name,
47 parent_alias,
48 table_alias,
49 join_type,
50 join_field,
51 nullable,
52 filtered_relation=None,
53 ):
54 # Join table
55 self.table_name = table_name
56 self.parent_alias = parent_alias
57 # Note: table_alias is not necessarily known at instantiation time.
58 self.table_alias = table_alias
59 # LOUTER or INNER
60 self.join_type = join_type
61 # A list of 2-tuples to use in the ON clause of the JOIN.
62 # Each 2-tuple will create one join condition in the ON clause.
63 self.join_cols = join_field.get_joining_columns()
64 # Along which field (or ForeignObjectRel in the reverse join case)
65 self.join_field = join_field
66 # Is this join nullabled?
67 self.nullable = nullable
68 self.filtered_relation = filtered_relation
70 def as_sql(self, compiler, connection):
71 """
72 Generate the full
73 LEFT OUTER JOIN sometable ON sometable.somecol = othertable.othercol, params
74 clause for this join.
75 """
76 join_conditions = []
77 params = []
78 qn = compiler.quote_name_unless_alias
79 qn2 = connection.ops.quote_name
81 # Add a join condition for each pair of joining columns.
82 for lhs_col, rhs_col in self.join_cols:
83 join_conditions.append(
84 "%s.%s = %s.%s"
85 % (
86 qn(self.parent_alias),
87 qn2(lhs_col),
88 qn(self.table_alias),
89 qn2(rhs_col),
90 )
91 )
93 # Add a single condition inside parentheses for whatever
94 # get_extra_restriction() returns.
95 extra_cond = self.join_field.get_extra_restriction(
96 self.table_alias, self.parent_alias
97 )
98 if extra_cond: 98 ↛ 99line 98 didn't jump to line 99, because the condition on line 98 was never true
99 extra_sql, extra_params = compiler.compile(extra_cond)
100 join_conditions.append("(%s)" % extra_sql)
101 params.extend(extra_params)
102 if self.filtered_relation: 102 ↛ 103line 102 didn't jump to line 103, because the condition on line 102 was never true
103 extra_sql, extra_params = compiler.compile(self.filtered_relation)
104 if extra_sql:
105 join_conditions.append("(%s)" % extra_sql)
106 params.extend(extra_params)
107 if not join_conditions: 107 ↛ 109line 107 didn't jump to line 109, because the condition on line 107 was never true
108 # This might be a rel on the other end of an actual declared field.
109 declared_field = getattr(self.join_field, "field", self.join_field)
110 raise ValueError(
111 "Join generated an empty ON clause. %s did not yield either "
112 "joining columns or extra restrictions." % declared_field.__class__
113 )
114 on_clause_sql = " AND ".join(join_conditions)
115 alias_str = (
116 "" if self.table_alias == self.table_name else (" %s" % self.table_alias)
117 )
118 sql = "%s %s%s ON (%s)" % (
119 self.join_type,
120 qn(self.table_name),
121 alias_str,
122 on_clause_sql,
123 )
124 return sql, params
126 def relabeled_clone(self, change_map):
127 new_parent_alias = change_map.get(self.parent_alias, self.parent_alias)
128 new_table_alias = change_map.get(self.table_alias, self.table_alias)
129 if self.filtered_relation is not None: 129 ↛ 130line 129 didn't jump to line 130, because the condition on line 129 was never true
130 filtered_relation = self.filtered_relation.clone()
131 filtered_relation.path = [
132 change_map.get(p, p) for p in self.filtered_relation.path
133 ]
134 else:
135 filtered_relation = None
136 return self.__class__(
137 self.table_name,
138 new_parent_alias,
139 new_table_alias,
140 self.join_type,
141 self.join_field,
142 self.nullable,
143 filtered_relation=filtered_relation,
144 )
146 @property
147 def identity(self):
148 return (
149 self.__class__,
150 self.table_name,
151 self.parent_alias,
152 self.join_field,
153 self.filtered_relation,
154 )
156 def __eq__(self, other):
157 if not isinstance(other, Join):
158 return NotImplemented
159 return self.identity == other.identity
161 def __hash__(self):
162 return hash(self.identity)
164 def equals(self, other):
165 # Ignore filtered_relation in equality check.
166 return self.identity[:-1] == other.identity[:-1]
168 def demote(self):
169 new = self.relabeled_clone({})
170 new.join_type = INNER
171 return new
173 def promote(self):
174 new = self.relabeled_clone({})
175 new.join_type = LOUTER
176 return new
179class BaseTable:
180 """
181 The BaseTable class is used for base table references in FROM clause. For
182 example, the SQL "foo" in
183 SELECT * FROM "foo" WHERE somecond
184 could be generated by this class.
185 """
187 join_type = None
188 parent_alias = None
189 filtered_relation = None
191 def __init__(self, table_name, alias):
192 self.table_name = table_name
193 self.table_alias = alias
195 def as_sql(self, compiler, connection):
196 alias_str = (
197 "" if self.table_alias == self.table_name else (" %s" % self.table_alias)
198 )
199 base_sql = compiler.quote_name_unless_alias(self.table_name)
200 return base_sql + alias_str, []
202 def relabeled_clone(self, change_map):
203 return self.__class__(
204 self.table_name, change_map.get(self.table_alias, self.table_alias)
205 )
207 @property
208 def identity(self):
209 return self.__class__, self.table_name, self.table_alias
211 def __eq__(self, other):
212 if not isinstance(other, BaseTable): 212 ↛ 214line 212 didn't jump to line 214, because the condition on line 212 was never false
213 return NotImplemented
214 return self.identity == other.identity
216 def __hash__(self):
217 return hash(self.identity)
219 def equals(self, other):
220 return self.identity == other.identity