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