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