refactor: decouple DefaultQueryProvider, search_utils, and Filter classes from SQLAlchemy

fastapi_admin_kit/admin/admin_database.py

@@ -6,6 +6,7 @@
 import re
 from typing import Any
 
+from fastapi_admin_kit.backends.sqlalchemy import SqlAlchemyDatabaseBackend
 from fastapi_admin_kit.config.database import DatabaseConfig
 
 logger = logging.getLogger(__name__)
@@ -21,7 +22,11 @@ def _validate_identifier(name: str, kind: str = "table") -> str:
 
 
 class AdminDatabase:
-    """Handles database setup, table creation, and role seeding."""
+    """Handles database setup, table creation, and role seeding.
+
+    Delegates engine creation, table creation, and auto-migration to
+    :class:`SqlAlchemyDatabaseBackend`.
+    """
 
     def __init__(
         self,
@@ -32,11 +37,12 @@ def __init__(
         self.engine = engine
         self.base = base
         self.database_config = database_config
+        self._backend = SqlAlchemyDatabaseBackend(
+            admin_database=self, database_config=database_config
+        )
 
     def _ensure_engine(self) -> Any:
-        """
-        Create the async engine from ``database_config`` if no engine is set.
-        """
+        """Create the async engine from ``database_config`` if no engine is set."""
         if self.engine is None and self.database_config is not None:
             self.engine = self.database_config.create_engine()
         return self.engine
@@ -108,9 +114,7 @@ def _auto_migrate_sync(self, metadata: Any) -> None:
                         conn.execute(sql)
 
     def _auto_migrate(self, sync_conn: Any, metadata: Any) -> None:
-        """
-        Add missing columns to existing tables (sync, called via run_sync).
-        """
+        """Add missing columns to existing tables (sync, called via run_sync)."""
         from sqlalchemy import inspect as sa_inspect
         from sqlalchemy import text
 

fastapi_admin_kit/admin/core.py

@@ -721,6 +721,16 @@ def _wire_app_state(self, app: FastAPI) -> None:
         app.state.admin_jinja_env = state.jinja_env
         # Unified signing-key source for sessions, CSRF, and JWT (see AdminState).
         app.state.admin_secret_key = state.secret_key
+        # Multi-ORM backend: store adapter class for views to access
+        from fastapi_admin_kit.backends.sqlalchemy import (
+            SqlAlchemyIntrospectionAdapter,
+            SqlAlchemyQueryAdapter,
+            SqlAlchemySessionAdapter,
+        )
+
+        app.state.admin_session_backend_class = SqlAlchemySessionAdapter
+        app.state.admin_query_adapter = SqlAlchemyQueryAdapter()
+        app.state.admin_introspection_adapter = SqlAlchemyIntrospectionAdapter()
 
         # Wire the password hasher to the User model
         from fastapi_admin_kit.auth.models import User

fastapi_admin_kit/backends/__init__.py

@@ -0,0 +1,57 @@
+"""Backend protocols and SQLAlchemy adapters for multi-ORM support.
+
+Protocols::
+
+    from fastapi_admin_kit.backends import (
+        IntrospectionBackend,
+        SessionBackend,
+        QueryBackend,
+        AuditBackend,
+        DatabaseBackend,
+    )
+
+SQLAlchemy adapters::
+
+    from fastapi_admin_kit.backends import (
+        SqlAlchemyIntrospectionAdapter,
+        SqlAlchemySessionAdapter,
+        SqlAlchemyQueryAdapter,
+        SqlAlchemyDatabaseBackend,
+    )
+"""
+
+from fastapi_admin_kit.backends.protocols import (
+    AuditBackend,
+    ColumnMetaType,
+    DatabaseBackend,
+    IntrospectionBackend,
+    QueryBackend,
+    QueryType,
+    RelationMetaType,
+    SessionBackend,
+    SessionType,
+)
+from fastapi_admin_kit.backends.sqlalchemy import (
+    SqlAlchemyDatabaseBackend,
+    SqlAlchemyIntrospectionAdapter,
+    SqlAlchemyQueryAdapter,
+    SqlAlchemySessionAdapter,
+)
+
+__all__ = [
+    # Protocols
+    "AuditBackend",
+    "ColumnMetaType",
+    "DatabaseBackend",
+    "IntrospectionBackend",
+    "QueryBackend",
+    "QueryType",
+    "RelationMetaType",
+    "SessionBackend",
+    "SessionType",
+    # SQLAlchemy adapters
+    "SqlAlchemyDatabaseBackend",
+    "SqlAlchemyIntrospectionAdapter",
+    "SqlAlchemyQueryAdapter",
+    "SqlAlchemySessionAdapter",
+]

fastapi_admin_kit/backends/protocols.py

@@ -0,0 +1,187 @@
+"""Protocol interfaces for multi-ORM backend support.
+
+These protocols define the contracts that all ORM backends (SQLAlchemy,
+MongoDB/ODM, future) must implement. They are the seam that decouples
+the rest of the codebase from any specific ORM.
+
+Use structural subtyping — any class that satisfies the protocol's
+method signatures is a valid implementation, no inheritance required.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any, Protocol, TypeVar, runtime_checkable
+
+from fastapi_admin_kit.types import ColumnMeta, RelationMeta
+
+ModelT = TypeVar("ModelT")
+ObjT = TypeVar("ObjT")
+
+# Type aliases — the rest of the codebase should reference these
+# instead of SQLAlchemy-specific types.
+QueryType = Any
+SessionType = Any
+ColumnMetaType = ColumnMeta
+RelationMetaType = RelationMeta
+
+
+@runtime_checkable
+class IntrospectionBackend(Protocol):
+    """Model introspection: reflect columns, relationships, PKs, and abstractness."""
+
+    def inspect_model(self, model: type) -> tuple[list[ColumnMeta], list[RelationMeta]]:
+        """Inspect a model and return its column and relationship metadata."""
+        ...
+
+    def get_pk_field(self, model: type) -> str | tuple[str, ...] | None:
+        """Return the primary key field name(s) for a model."""
+        ...
+
+    def cast_pk_value(self, model: type, value: Any) -> Any:
+        """Cast a string PK value to the correct Python type for the model."""
+        ...
+
+    def is_abstract(self, model: type) -> bool:
+        """Return True if the model is abstract and should be skipped."""
+        ...
+
+    def get_relationship_names(self, model: type) -> set[str]:
+        """Return the set of relationship key names on a model."""
+        ...
+
+    def get_relationship(self, model: type, name: str) -> Any:
+        """Return a single relationship descriptor by name, or None."""
+        ...
+
+    def get_column_type_name(self, model: type, field_name: str) -> str | None:
+        """Return the SQLAlchemy type class name for a column, or None."""
+        ...
+
+    def get_column_attr(self, model: type, field_name: str) -> Any:
+        """Return the column attribute for a field name, or None."""
+        ...
+
+    def get_pk_columns(self, model: type) -> list[Any]:
+        """Return the primary key column(s) for a model."""
+        ...
+
+
+@runtime_checkable
+class SessionBackend(Protocol):
+    """Data access: per-request session lifecycle."""
+
+    def get(self, model: type[ModelT], pk: Any) -> ModelT | None:
+        """Fetch a single object by primary key."""
+        ...
+
+    def add(self, obj: Any) -> None:
+        """Stage an object for insertion."""
+        ...
+
+    def flush(self) -> None:
+        """Flush pending changes to the DB without committing."""
+        ...
+
+    def delete(self, obj: Any) -> None:
+        """Mark an object for deletion."""
+        ...
+
+    def refresh(self, obj: Any, attributes: Sequence[str] | None = None) -> None:
+        """Re-read object attributes from the DB."""
+        ...
+
+    def execute(self, query: QueryType) -> Any:
+        """Execute a query object and return the result."""
+        ...
+
+    def commit(self) -> None:
+        """Persist all pending changes."""
+        ...
+
+    def rollback(self) -> None:
+        """Discard all pending changes."""
+        ...
+
+
+@runtime_checkable
+class QueryBackend(Protocol):
+    """Chainable query building: select, filter, sort, join, paginate."""
+
+    def select(self, model: type[ModelT]) -> QueryType:
+        """Start a new query for the given model."""
+        ...
+
+    def where(self, query: QueryType, *conditions: Any) -> QueryType:
+        """Add WHERE conditions to a query."""
+        ...
+
+    def order_by(self, query: QueryType, *columns: Any) -> QueryType:
+        """Add ORDER BY clauses to a query."""
+        ...
+
+    def limit(self, query: QueryType, n: int) -> QueryType:
+        """Limit the result set to *n* rows."""
+        ...
+
+    def offset(self, query: QueryType, n: int) -> QueryType:
+        """Skip the first *n* rows of the result set."""
+        ...
+
+    def join(self, query: QueryType, related: type, on: Any | None = None) -> QueryType:
+        """Join a related model onto the query."""
+        ...
+
+    def distinct(self, query: QueryType) -> QueryType:
+        """Add DISTINCT to the query."""
+        ...
+
+    def count(self, query: QueryType) -> int:
+        """Execute the query and return the total row count."""
+        ...
+
+    def options(self, query: QueryType, *opts: Any) -> QueryType:
+        """Add eager-load options (joinedload, selectinload, etc.)."""
+        ...
+
+    def ilike(self, column: Any, pattern: str) -> Any:
+        """Apply case-insensitive LIKE to a column, returning a boolean clause."""
+        ...
+
+    def or_(self, *clauses: Any) -> Any:
+        """Compose multiple boolean clauses with OR."""
+        ...
+
+
+@runtime_checkable
+class AuditBackend(Protocol):
+    """Change tracking: attach listeners, snapshot, and diff objects."""
+
+    def attach_listeners(self, session_factory: Any, registry: dict[str, Any]) -> None:
+        """Register change-tracking listeners on the session factory."""
+        ...
+
+    def snapshot(self, obj: Any) -> dict[str, Any]:
+        """Capture a serialisable snapshot of the object's current state."""
+        ...
+
+    def compute_diff(self, before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
+        """Return a dict of {field: (old_value, new_value)} for changed fields."""
+        ...
+
+
+@runtime_checkable
+class DatabaseBackend(Protocol):
+    """Connection lifecycle: create engine, run DDL, auto-migrate."""
+
+    def create_connection(self) -> Any:
+        """Create and return a new database connection or engine."""
+        ...
+
+    def create_tables(self, connection: Any, metadata: Any) -> None:
+        """Issue DDL to create all tables defined in *metadata*."""
+        ...
+
+    def auto_migrate(self, connection: Any, metadata: Any) -> None:
+        """Detect schema drift and apply migrations automatically."""
+        ...