Type annotations for plain-models

Author: davegaeddert

plain-models/plain/models/aggregates.py

@@ -3,7 +3,14 @@
 from typing import TYPE_CHECKING, Any
 
 from plain.models.exceptions import FieldError, FullResultSet
-from plain.models.expressions import Case, Func, Star, Value, When
+from plain.models.expressions import (
+    Case,
+    Func,
+    ResolvableExpression,
+    Star,
+    Value,
+    When,
+)
 from plain.models.fields import IntegerField
 from plain.models.functions.comparison import Coalesce
 from plain.models.functions.mixins import (
@@ -99,7 +106,7 @@ def resolve_expression(
                     )
         if (default := c.default) is None:
             return c
-        if hasattr(default, "resolve_expression"):
+        if isinstance(default, ResolvableExpression):
             default = default.resolve_expression(query, allow_joins, reuse, summarize)
             if default._output_field_or_none is None:
                 default.output_field = c._output_field_or_none
@@ -132,7 +139,7 @@ def as_sql(
         if self.filter is not None:
             if connection.features.supports_aggregate_filter_clause:
                 try:
-                    filter_sql, filter_params = self.filter.as_sql(compiler, connection)
+                    filter_sql, filter_params = self.filter.as_sql(compiler, connection)  # type: ignore[possibly-missing-attribute]
                 except FullResultSet:
                     pass
                 else:

plain-models/plain/models/backends/base/base.py

@@ -48,6 +48,12 @@ class BaseDatabaseWrapper(ABC):
     data_types_suffix: dict[str, str] = {}
     # Mapping of Field objects to their SQL for CHECK constraints.
     data_type_check_constraints: dict[str, str] = {}
+    # Mapping of lookup operators to SQL templates (defined on backend subclasses)
+    operators: dict[str, str]
+    # Mapping of pattern lookup operators to SQL templates using str.format syntax (defined on backend subclasses)
+    pattern_ops: dict[str, str]
+    # SQL template for escaping patterns in LIKE queries using str.format syntax (defined on backend subclasses)
+    pattern_esc: str
     # Instance attributes - always assigned in __init__
     ops: BaseDatabaseOperations
     client: BaseDatabaseClient
@@ -268,7 +274,7 @@ def ensure_connection(self) -> None:
 
     # ##### Backend-specific wrappers for PEP-249 connection methods #####
 
-    def _prepare_cursor(self, cursor: Any) -> utils.CursorWrapper:
+    def _prepare_cursor(self, cursor: utils.DBAPICursor) -> utils.CursorWrapper:
         """
         Validate the connection is usable and perform database cursor wrapping.
         """
@@ -460,7 +466,7 @@ def set_autocommit(
         )
 
         if start_transaction_under_autocommit:
-            self._start_transaction_under_autocommit()
+            self._start_transaction_under_autocommit()  # type: ignore[attr-defined]
         elif autocommit:
             self._set_autocommit(autocommit)
         else:
@@ -629,11 +635,11 @@ def chunked_cursor(self) -> utils.CursorWrapper:
         """
         return self.cursor()
 
-    def make_debug_cursor(self, cursor: Any) -> utils.CursorDebugWrapper:
+    def make_debug_cursor(self, cursor: utils.DBAPICursor) -> utils.CursorDebugWrapper:
         """Create a cursor that logs all queries in self.queries_log."""
         return utils.CursorDebugWrapper(cursor, self)
 
-    def make_cursor(self, cursor: Any) -> utils.CursorWrapper:
+    def make_cursor(self, cursor: utils.DBAPICursor) -> utils.CursorWrapper:
         """Create a cursor without debug logging."""
         return utils.CursorWrapper(cursor, self)
 

plain-models/plain/models/backends/base/introspection.py

@@ -4,6 +4,8 @@
 from collections.abc import Generator
 from typing import TYPE_CHECKING, Any, NamedTuple
 
+from plain.models.backends.utils import CursorWrapper
+
 if TYPE_CHECKING:
     from plain.models.backends.base.base import BaseDatabaseWrapper
 
@@ -55,15 +57,17 @@ def identifier_converter(self, name: str) -> str:
         """
         return name
 
-    def table_names(self, cursor: Any = None, include_views: bool = False) -> list[str]:
+    def table_names(
+        self, cursor: CursorWrapper | None = None, include_views: bool = False
+    ) -> list[str]:
         """
         Return a list of names of all tables that exist in the database.
         Sort the returned table list by Python's default sorting. Do NOT use
         the database's ORDER BY here to avoid subtle differences in sorting
         order between databases.
         """
 
-        def get_names(cursor: Any) -> list[str]:
+        def get_names(cursor: CursorWrapper) -> list[str]:
             return sorted(
                 ti.name
                 for ti in self.get_table_list(cursor)
@@ -76,15 +80,17 @@ def get_names(cursor: Any) -> list[str]:
         return get_names(cursor)
 
     @abstractmethod
-    def get_table_list(self, cursor: Any) -> list[TableInfo]:
+    def get_table_list(self, cursor: CursorWrapper) -> list[TableInfo]:
         """
         Return an unsorted list of TableInfo named tuples of all tables and
         views that exist in the database.
         """
         ...
 
     @abstractmethod
-    def get_table_description(self, cursor: Any, table_name: str) -> list[FieldInfo]:
+    def get_table_description(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> list[FieldInfo]:
         """
         Return a description of the table with the DB-API cursor.description
         interface.
@@ -146,7 +152,7 @@ def sequence_list(self) -> list[dict[str, Any]]:
 
     @abstractmethod
     def get_sequences(
-        self, cursor: Any, table_name: str, table_fields: tuple[Any, ...] = ()
+        self, cursor: CursorWrapper, table_name: str, table_fields: tuple[Any, ...] = ()
     ) -> list[dict[str, Any]]:
         """
         Return a list of introspected sequences for table_name. Each sequence
@@ -156,21 +162,27 @@ def get_sequences(
         ...
 
     @abstractmethod
-    def get_relations(self, cursor: Any, table_name: str) -> dict[str, tuple[str, str]]:
+    def get_relations(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> dict[str, tuple[str, str]]:
         """
         Return a dictionary of {field_name: (field_name_other_table, other_table)}
         representing all foreign keys in the given table.
         """
         ...
 
-    def get_primary_key_column(self, cursor: Any, table_name: str) -> str | None:
+    def get_primary_key_column(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> str | None:
         """
         Return the name of the primary key column for the given table.
         """
         columns = self.get_primary_key_columns(cursor, table_name)
         return columns[0] if columns else None
 
-    def get_primary_key_columns(self, cursor: Any, table_name: str) -> list[str] | None:
+    def get_primary_key_columns(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> list[str] | None:
         """Return a list of primary key columns for the given table."""
         for constraint in self.get_constraints(cursor, table_name).values():
             if constraint["primary_key"]:
@@ -179,7 +191,7 @@ def get_primary_key_columns(self, cursor: Any, table_name: str) -> list[str] | N
 
     @abstractmethod
     def get_constraints(
-        self, cursor: Any, table_name: str
+        self, cursor: CursorWrapper, table_name: str
     ) -> dict[str, dict[str, Any]]:
         """
         Retrieve any constraints or keys (unique, pk, fk, check, index)

plain-models/plain/models/backends/base/operations.py

@@ -11,7 +11,9 @@
 import sqlparse
 
 from plain.models.backends import utils
+from plain.models.backends.utils import CursorWrapper
 from plain.models.db import NotSupportedError
+from plain.models.expressions import ResolvableExpression
 from plain.utils import timezone
 from plain.utils.encoding import force_str
 
@@ -223,7 +225,9 @@ def distinct_sql(
         else:
             return ["DISTINCT"], []
 
-    def fetch_returned_insert_columns(self, cursor: Any, returning_params: Any) -> Any:
+    def fetch_returned_insert_columns(
+        self, cursor: CursorWrapper, returning_params: Any
+    ) -> Any:
         """
         Given a cursor object that has just performed an INSERT...RETURNING
         statement into a table, return the newly created data.
@@ -287,7 +291,7 @@ def limit_offset_sql(self, low_mark: int | None, high_mark: int | None) -> str:
 
     def last_executed_query(
         self,
-        cursor: Any,
+        cursor: CursorWrapper,
         sql: str,
         params: list[Any] | tuple[Any, ...] | dict[str, Any] | None,
     ) -> str:
@@ -315,7 +319,9 @@ def to_string(s: Any) -> str:
 
         return f"QUERY = {sql!r} - PARAMS = {u_params!r}"
 
-    def last_insert_id(self, cursor: Any, table_name: str, pk_name: str) -> int:
+    def last_insert_id(
+        self, cursor: CursorWrapper, table_name: str, pk_name: str
+    ) -> int:
         """
         Given a cursor object that has just performed an INSERT statement into
         a table that has an auto-incrementing ID, return the newly created ID.
@@ -396,7 +402,7 @@ def bulk_insert_sql(
         ...
 
     @abstractmethod
-    def fetch_returned_insert_rows(self, cursor: Any) -> list[Any]:
+    def fetch_returned_insert_rows(self, cursor: CursorWrapper) -> list[Any]:
         """
         Given a cursor object that has just performed an INSERT...RETURNING
         statement into a table, return the list of returned data.
@@ -521,7 +527,7 @@ def adapt_datetimefield_value(
         if value is None:
             return None
         # Expression values are adapted by the database.
-        if hasattr(value, "resolve_expression"):
+        if isinstance(value, ResolvableExpression):
             return value
 
         return str(value)
@@ -536,7 +542,7 @@ def adapt_timefield_value(
         if value is None:
             return None
         # Expression values are adapted by the database.
-        if hasattr(value, "resolve_expression"):
+        if isinstance(value, ResolvableExpression):
             return value
 
         if timezone.is_aware(value):

plain-models/plain/models/backends/mysql/base.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from functools import cached_property
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import MySQLdb as Database
 from MySQLdb.constants import CLIENT, FIELD_TYPE
@@ -19,11 +19,14 @@
 from plain.models.db import IntegrityError
 from plain.utils.regex_helper import _lazy_re_compile
 
-# With mysqlclient stubs, we can now type the connection
-try:
+# Type checkers always see the proper type; runtime falls back to Any if needed
+if TYPE_CHECKING:
     from MySQLdb.connections import Connection as MySQLConnection
-except ImportError:
-    MySQLConnection: type[Any] = Any  # type: ignore[misc]
+else:
+    try:
+        from MySQLdb.connections import Connection as MySQLConnection
+    except ImportError:
+        MySQLConnection = Any  # type: ignore[misc]
 
 from .client import DatabaseClient
 from .creation import DatabaseCreation

plain-models/plain/models/backends/mysql/introspection.py

@@ -8,6 +8,7 @@
 from plain.models.backends.base.introspection import (
     BaseDatabaseIntrospection,
 )
+from plain.models.backends.utils import CursorWrapper
 from plain.models.indexes import Index
 from plain.utils.datastructures import OrderedSet
 
@@ -106,7 +107,7 @@ def get_field_type(self, data_type: Any, description: Any) -> str:
             return "JSONField"
         return field_type
 
-    def get_table_list(self, cursor: Any) -> list[TableInfo]:
+    def get_table_list(self, cursor: CursorWrapper) -> list[TableInfo]:
         """Return a list of table and view names in the current database."""
         cursor.execute(
             """
@@ -119,11 +120,13 @@ def get_table_list(self, cursor: Any) -> list[TableInfo]:
             """
         )
         return [
-            TableInfo(row[0], {"BASE TABLE": "t", "VIEW": "v"}.get(row[1]), row[2])
+            TableInfo(row[0], {"BASE TABLE": "t", "VIEW": "v"}.get(row[1], "t"), row[2])
             for row in cursor.fetchall()
         ]
 
-    def get_table_description(self, cursor: Any, table_name: str) -> list[FieldInfo]:
+    def get_table_description(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> list[FieldInfo]:
         """
         Return a description of the table with the DB-API cursor.description
         interface."
@@ -216,15 +219,17 @@ def to_int(i: Any) -> Any:
         return fields
 
     def get_sequences(
-        self, cursor: Any, table_name: str, table_fields: tuple[Any, ...] = ()
+        self, cursor: CursorWrapper, table_name: str, table_fields: tuple[Any, ...] = ()
     ) -> list[dict[str, Any]]:
         for field_info in self.get_table_description(cursor, table_name):
             if "auto_increment" in field_info.extra:
                 # MySQL allows only one auto-increment column per table.
                 return [{"table": table_name, "column": field_info.name}]
         return []
 
-    def get_relations(self, cursor: Any, table_name: str) -> dict[str, tuple[str, str]]:
+    def get_relations(
+        self, cursor: CursorWrapper, table_name: str
+    ) -> dict[str, tuple[str, str]]:
         """
         Return a dictionary of {field_name: (field_name_other_table, other_table)}
         representing all foreign keys in the given table.
@@ -245,7 +250,7 @@ def get_relations(self, cursor: Any, table_name: str) -> dict[str, tuple[str, st
             for field_name, other_field, other_table in cursor.fetchall()
         }
 
-    def get_storage_engine(self, cursor: Any, table_name: str) -> str:
+    def get_storage_engine(self, cursor: CursorWrapper, table_name: str) -> str:
         """
         Retrieve the storage engine for a given table. Return the default
         storage engine if the table doesn't exist.
@@ -281,7 +286,7 @@ def _parse_constraint_columns(
         return check_columns
 
     def get_constraints(
-        self, cursor: Any, table_name: str
+        self, cursor: CursorWrapper, table_name: str
     ) -> dict[str, dict[str, Any]]:
         """
         Retrieve any constraints or keys (unique, pk, fk, check, index) across

plain-models/plain/models/backends/mysql/operations.py

@@ -6,9 +6,9 @@
 from typing import TYPE_CHECKING, Any
 
 from plain.models.backends.base.operations import BaseDatabaseOperations
-from plain.models.backends.utils import split_tzname_delta
+from plain.models.backends.utils import CursorWrapper, split_tzname_delta
 from plain.models.constants import OnConflict
-from plain.models.expressions import Exists, ExpressionWrapper
+from plain.models.expressions import Exists, ExpressionWrapper, ResolvableExpression
 from plain.models.lookups import Lookup
 from plain.utils import timezone
 from plain.utils.encoding import force_str
@@ -191,7 +191,7 @@ def time_trunc_sql(
         else:
             return f"TIME({sql})", params
 
-    def fetch_returned_insert_rows(self, cursor: Any) -> list[Any]:
+    def fetch_returned_insert_rows(self, cursor: CursorWrapper) -> list[Any]:
         """
         Given a cursor object that has just performed an INSERT...RETURNING
         statement into a table, return the tuple of returned data.
@@ -217,7 +217,9 @@ def adapt_decimalfield_value(
     ) -> Any:
         return value
 
-    def last_executed_query(self, cursor: Any, sql: str, params: Any) -> str | None:
+    def last_executed_query(
+        self, cursor: CursorWrapper, sql: str, params: Any
+    ) -> str | None:
         # With MySQLdb, cursor objects have an (undocumented) "_executed"
         # attribute where the exact query sent to the database is saved.
         # See MySQLdb/cursors.py in the source distribution.
@@ -260,7 +262,7 @@ def adapt_datetimefield_value(
             return None
 
         # Expression values are adapted by the database.
-        if hasattr(value, "resolve_expression"):
+        if isinstance(value, ResolvableExpression):
             return value
 
         # MySQL doesn't support tz-aware datetimes
@@ -275,7 +277,7 @@ def adapt_timefield_value(
             return None
 
         # Expression values are adapted by the database.
-        if hasattr(value, "resolve_expression"):
+        if isinstance(value, ResolvableExpression):
             return value
 
         # MySQL doesn't support tz-aware times