1from __future__ import annotations
 2
 3import operator
 4from functools import cached_property
 5from typing import TYPE_CHECKING
 6
 7from plain.models import transaction
 8from plain.models.backends.base.features import BaseDatabaseFeatures
 9from plain.models.db import OperationalError
10
11from .base import Database
12
13if TYPE_CHECKING:
14    from plain.models.backends.sqlite3.base import SQLiteDatabaseWrapper
15
16
17class DatabaseFeatures(BaseDatabaseFeatures):
18    # Type checker hint: connection is always SQLiteDatabaseWrapper in this class
19    connection: SQLiteDatabaseWrapper
20
21    minimum_database_version = (3, 21)
22    max_query_params = 999
23    supports_transactions = True
24    can_rollback_ddl = True
25    requires_literal_defaults = True
26    supports_temporal_subtraction = True
27    ignores_table_name_case = True
28    # Is "ALTER TABLE ... RENAME COLUMN" supported?
29    can_alter_table_rename_column = Database.sqlite_version_info >= (3, 25, 0)
30    # Is "ALTER TABLE ... DROP COLUMN" supported?
31    can_alter_table_drop_column = Database.sqlite_version_info >= (3, 35, 5)
32    can_defer_constraint_checks = True
33    supports_over_clause = Database.sqlite_version_info >= (3, 25, 0)
34    supports_aggregate_filter_clause = Database.sqlite_version_info >= (3, 30, 1)
35    supports_order_by_nulls_modifier = Database.sqlite_version_info >= (3, 30, 0)
36    order_by_nulls_first = True
37    supports_json_field_contains = False
38    supports_update_conflicts = Database.sqlite_version_info >= (3, 24, 0)
39    supports_update_conflicts_with_target = supports_update_conflicts
40    supports_unlimited_charfield = True
41
42    @cached_property
43    def supports_atomic_references_rename(self) -> bool:
44        return Database.sqlite_version_info >= (3, 26, 0)
45
46    @cached_property
47    def supports_json_field(self) -> bool:
48        with self.connection.cursor() as cursor:
49            try:
50                with transaction.atomic():
51                    cursor.execute('SELECT JSON(\'{"a": "b"}\')')
52            except OperationalError:
53                return False
54        return True
55
56    can_introspect_json_field = property(operator.attrgetter("supports_json_field"))
57    has_json_object_function = property(operator.attrgetter("supports_json_field"))
58
59    @cached_property
60    def can_return_columns_from_insert(self) -> bool:
61        return Database.sqlite_version_info >= (3, 35)
62
63    can_return_rows_from_bulk_insert = property(
64        operator.attrgetter("can_return_columns_from_insert")
65    )