1from __future__ import annotations
2
3import operator
4from functools import cached_property
5
6from plain.models.backends.base.features import BaseDatabaseFeatures
7
8
9class DatabaseFeatures(BaseDatabaseFeatures):
10 empty_fetchmany_value = ()
11 allows_group_by_selected_pks = True
12 related_fields_match_type = True
13 has_select_for_update = True
14 supports_comments = True
15 supports_comments_inline = True
16 supports_temporal_subtraction = True
17 supports_slicing_ordering_in_compound = True
18 supports_update_conflicts = True
19
20 # Neither MySQL nor MariaDB support partial indexes.
21 supports_partial_indexes = False
22 # COLLATE must be wrapped in parentheses because MySQL treats COLLATE as an
23 # indexed expression.
24 collate_as_index_expression = True
25
26 supports_order_by_nulls_modifier = False
27 order_by_nulls_first = True
28 supports_logical_xor = True
29
30 @cached_property
31 def minimum_database_version(self) -> tuple[int, ...]:
32 if self.connection.mysql_is_mariadb:
33 return (10, 4)
34 else:
35 return (8,)
36
37 @cached_property
38 def _mysql_storage_engine(self) -> str:
39 "Internal method used in Plain tests. Don't rely on this from your code"
40 return self.connection.mysql_server_data["default_storage_engine"]
41
42 @cached_property
43 def allows_auto_pk_0(self) -> bool:
44 """
45 Autoincrement primary key can be set to 0 if it doesn't generate new
46 autoincrement values.
47 """
48 return "NO_AUTO_VALUE_ON_ZERO" in self.connection.sql_mode
49
50 @cached_property
51 def update_can_self_select(self) -> bool:
52 return self.connection.mysql_is_mariadb and self.connection.mysql_version >= (
53 10,
54 3,
55 2,
56 )
57
58 @cached_property
59 def can_return_columns_from_insert(self) -> bool:
60 return self.connection.mysql_is_mariadb and self.connection.mysql_version >= (
61 10,
62 5,
63 0,
64 )
65
66 can_return_rows_from_bulk_insert = property(
67 operator.attrgetter("can_return_columns_from_insert")
68 )
69
70 @cached_property
71 def has_zoneinfo_database(self) -> bool:
72 return self.connection.mysql_server_data["has_zoneinfo_database"]
73
74 @cached_property
75 def is_sql_auto_is_null_enabled(self) -> bool:
76 return self.connection.mysql_server_data["sql_auto_is_null"]
77
78 @cached_property
79 def supports_over_clause(self) -> bool:
80 if self.connection.mysql_is_mariadb:
81 return True
82 return self.connection.mysql_version >= (8, 0, 2)
83
84 @cached_property
85 def supports_column_check_constraints(self) -> bool:
86 if self.connection.mysql_is_mariadb:
87 return True
88 return self.connection.mysql_version >= (8, 0, 16)
89
90 supports_table_check_constraints = property(
91 operator.attrgetter("supports_column_check_constraints")
92 )
93
94 @cached_property
95 def can_introspect_check_constraints(self) -> bool:
96 if self.connection.mysql_is_mariadb:
97 return True
98 return self.connection.mysql_version >= (8, 0, 16)
99
100 @cached_property
101 def has_select_for_update_skip_locked(self) -> bool:
102 if self.connection.mysql_is_mariadb:
103 return self.connection.mysql_version >= (10, 6)
104 return self.connection.mysql_version >= (8, 0, 1)
105
106 @cached_property
107 def has_select_for_update_nowait(self) -> bool:
108 if self.connection.mysql_is_mariadb:
109 return True
110 return self.connection.mysql_version >= (8, 0, 1)
111
112 @cached_property
113 def has_select_for_update_of(self) -> bool:
114 return (
115 not self.connection.mysql_is_mariadb
116 and self.connection.mysql_version >= (8, 0, 1)
117 )
118
119 @cached_property
120 def supports_explain_analyze(self) -> bool:
121 return self.connection.mysql_is_mariadb or self.connection.mysql_version >= (
122 8,
123 0,
124 18,
125 )
126
127 @cached_property
128 def supported_explain_formats(self) -> set[str]:
129 # Alias MySQL's TRADITIONAL to TEXT for consistency with other
130 # backends.
131 formats = {"JSON", "TEXT", "TRADITIONAL"}
132 if not self.connection.mysql_is_mariadb and self.connection.mysql_version >= (
133 8,
134 0,
135 16,
136 ):
137 formats.add("TREE")
138 return formats
139
140 @cached_property
141 def supports_transactions(self) -> bool:
142 """
143 All storage engines except MyISAM support transactions.
144 """
145 return self._mysql_storage_engine != "MyISAM"
146
147 uses_savepoints = property(operator.attrgetter("supports_transactions"))
148
149 @cached_property
150 def ignores_table_name_case(self) -> bool:
151 return self.connection.mysql_server_data["lower_case_table_names"]
152
153 @cached_property
154 def can_introspect_json_field(self) -> bool:
155 if self.connection.mysql_is_mariadb:
156 return self.can_introspect_check_constraints
157 return True
158
159 @cached_property
160 def supports_index_column_ordering(self) -> bool:
161 if self._mysql_storage_engine != "InnoDB":
162 return False
163 if self.connection.mysql_is_mariadb:
164 return self.connection.mysql_version >= (10, 8)
165 return self.connection.mysql_version >= (8, 0, 1)
166
167 @cached_property
168 def supports_expression_indexes(self) -> bool:
169 return (
170 not self.connection.mysql_is_mariadb
171 and self._mysql_storage_engine != "MyISAM"
172 and self.connection.mysql_version >= (8, 0, 13)
173 )
174
175 @cached_property
176 def supports_select_intersection(self) -> bool:
177 is_mariadb = self.connection.mysql_is_mariadb
178 return is_mariadb or self.connection.mysql_version >= (8, 0, 31)
179
180 supports_select_difference = property(
181 operator.attrgetter("supports_select_intersection")
182 )
183
184 @cached_property
185 def can_rename_index(self) -> bool:
186 if self.connection.mysql_is_mariadb:
187 return self.connection.mysql_version >= (10, 5, 2)
188 return True