Lots of work for reports support.

crudkit/core/service.py

@@ -1,12 +1,34 @@
-from typing import Type, TypeVar, Generic, Optional
-from sqlalchemy.orm import Load, Session, raiseload, with_polymorphic
+from typing import Any, Type, TypeVar, Generic, Optional, Protocol, runtime_checkable, cast
+from sqlalchemy.orm import Load, Session, raiseload, with_polymorphic, Mapper
 from sqlalchemy.orm.attributes import InstrumentedAttribute
+from sqlalchemy.orm.util import AliasedClass
+from sqlalchemy.engine import Engine, Connection
 from sqlalchemy import inspect, text
 from crudkit.core.base import Version
 from crudkit.core.spec import CRUDSpec
 from crudkit.backend import BackendInfo, make_backend_info
 
-T = TypeVar("T")
+@runtime_checkable
+class _HasID(Protocol):
+    id: int
+
+@runtime_checkable
+class _HasTable(Protocol):
+    __table__: Any
+
+@runtime_checkable
+class _HasADict(Protocol):
+    def as_dict(self) -> dict: ...
+
+@runtime_checkable
+class _SoftDeletable(Protocol):
+    is_deleted: bool
+
+class _CRUDModelProto(_HasID, _HasTable, _HasADict, Protocol):
+    """Minimal surface that our CRUD service relies on. Soft-delete is optional."""
+    pass
+
+T = TypeVar("T", bound=_CRUDModelProto)
 
 def _is_truthy(val):
     return str(val).lower() in ('1', 'true', 'yes', 'on')
@@ -25,7 +47,9 @@ class CRUDService(Generic[T]):
         self.polymorphic = polymorphic
         self.supports_soft_delete = hasattr(model, 'is_deleted')
         # Cache backend info once. If not provided, derive from session bind.
-        self.backend = backend or make_backend_info(self.session.get_bind())
+        bind = self.session.get_bind()
+        eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
+        self.backend = backend or make_backend_info(eng)
 
     def get_query(self):
         if self.polymorphic:
@@ -35,7 +59,7 @@ class CRUDService(Generic[T]):
 
     # Helper: default ORDER BY for MSSQL when paginating without explicit order
     def _default_order_by(self, root_alias):
-        mapper = inspect(self.model)
+        mapper: Mapper[Any] = cast(Mapper[Any], inspect(self.model))
         cols = []
         for col in mapper.primary_key:
             try:
@@ -64,11 +88,9 @@ class CRUDService(Generic[T]):
         spec.parse_includes()
 
         for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-            query = query.join(
-                target_alias,
-                relationship_attr.of_type(target_alias),
-                isouter=True
-            )
+            rel_attr = cast(InstrumentedAttribute, relationship_attr)
+            target = cast(Any, target_alias)
+            query = query.join(target, rel_attr.of_type(target), isouter=True)
 
         if params:
             root_fields, rel_field_names, root_field_names = spec.parse_fields()
@@ -123,11 +145,9 @@ class CRUDService(Generic[T]):
             spec.parse_includes()
 
             for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
-                query = query.join(
-                    target_alias,
-                    relationship_attr.of_type(target_alias),
-                    isouter=True
-                )
+                rel_attr = cast(InstrumentedAttribute, relationship_attr)
+                target = cast(Any, target_alias)
+                query = query.join(target, rel_attr.of_type(target), isouter=True)
 
             if params:
                 root_fields, rel_field_names, root_field_names = spec.parse_fields()
@@ -206,7 +226,8 @@ class CRUDService(Generic[T]):
         if hard or not self.supports_soft_delete:
             self.session.delete(obj)
         else:
-            obj.is_deleted = True
+            soft = cast(_SoftDeletable, obj)
+            soft.is_deleted = True
         self.session.commit()
         self._log_version("delete", obj, actor)
         return obj