feat: add deferred_column_property for safe deferred loading in SQLModel

Author: 50Bytes-dev

sqlmodel/__init__.py

@@ -114,6 +114,7 @@
 from sqlalchemy.types import Uuid as Uuid
 
 # From SQLModel, modifications of SQLAlchemy or equivalents of Pydantic
+from .deferred_column import deferred_column_property as deferred_column_property
 from .main import Field as Field
 from .main import Relationship as Relationship
 from .main import SQLModel as SQLModel

sqlmodel/deferred_column.py

@@ -0,0 +1,113 @@
+"""
+Deferred Column Property implementation for SQLModel.
+
+This module provides deferred_column_property function that creates column properties
+with safe deferred loading - returning fallback values instead of raising DetachedInstanceError.
+"""
+
+from typing import Any
+
+from sqlalchemy.orm import ColumnProperty
+from sqlalchemy.orm.strategies import DeferredColumnLoader, _state_session
+
+
+class SafeDeferredColumnLoader(DeferredColumnLoader):
+    """
+    A custom deferred column loader that returns a fallback value instead of
+    raising DetachedInstanceError when the session is detached.
+    """
+
+    def __init__(self, parent, strategy_key, fallback_value=None):
+        super().__init__(parent, strategy_key)
+        self.fallback_value = fallback_value
+
+    def _load_for_state(self, state, passive):
+        """
+        Override the default behavior to return fallback value instead of raising
+        DetachedInstanceError when session is None.
+        """
+        from sqlalchemy.orm import LoaderCallableStatus
+
+        if not state.key:
+            return LoaderCallableStatus.ATTR_EMPTY
+
+        # Check passive flags
+        from sqlalchemy.orm import PassiveFlag
+
+        if not passive & PassiveFlag.SQL_OK:
+            return LoaderCallableStatus.PASSIVE_NO_RESULT
+
+        # Check if we have a session before attempting to load
+        session = _state_session(state)
+        if session is None:
+            # No session available, return fallback value directly
+            return self.fallback_value
+
+        # We have a session, use the parent implementation
+        return super()._load_for_state(state, passive)
+
+
+class SafeColumnProperty(ColumnProperty):
+    """
+    Custom ColumnProperty that uses SafeDeferredColumnLoader for deferred loading.
+    """
+
+    def __init__(self, *args, fallback_value=None, **kwargs):
+        self.fallback_value = fallback_value
+        super().__init__(*args, **kwargs)
+
+    def do_init(self):
+        """Override to set our custom strategy after parent initialization."""
+        super().do_init()
+
+        # Replace the strategy with our safe version if it's deferred
+        if self.deferred and hasattr(self, "strategy"):
+            # Create our safe loader with the same parameters as the original
+            original_strategy = self.strategy
+
+            # Create new strategy with proper initialization
+            safe_strategy = SafeDeferredColumnLoader(
+                parent=self,  # The ColumnProperty itself
+                strategy_key=original_strategy.strategy_key,
+                fallback_value=self.fallback_value,
+            )
+
+            self.strategy = safe_strategy
+
+
+def deferred_column_property(
+    expression: Any, *, fallback_value: Any, deferred: bool = True, **kwargs: Any
+) -> SafeColumnProperty:
+    """
+    Create a deferred column property that returns a fallback value instead of raising
+    DetachedInstanceError when accessed on a detached instance.
+
+    This function behaves exactly like SQLAlchemy's column_property, but with safe
+    deferred loading behavior when the session is detached.
+
+    Args:
+        expression: The SQL expression for the column property
+        fallback_value: Value to return when property cannot be loaded (required)
+        deferred: Whether to defer loading the property (defaults to True)
+        **kwargs: Additional arguments passed to column_property
+
+    Returns:
+        A SafeColumnProperty instance
+
+    Example:
+        ```python
+        class User(SQLModel, table=True):
+            id: Optional[int] = Field(default=None, primary_key=True)
+            user_id: Optional[int] = None
+
+            @classmethod
+            def __declare_last__(cls):
+                cls.computed_value = deferred_column_property(
+                    cls.__table__.c.user_id * 2,
+                    fallback_value=0,
+                    deferred=True
+                )
+        ```
+    """
+    kwargs["deferred"] = deferred
+    return SafeColumnProperty(expression, fallback_value=fallback_value, **kwargs)

sqlmodel/main.py

@@ -149,17 +149,15 @@ def __init__(self, default: Any = Undefined, **kwargs: Any) -> None:
                 )
             if primary_key is not Undefined:
                 raise RuntimeError(
-                    "Passing primary_key is not supported when "
-                    "also passing a sa_column"
+                    "Passing primary_key is not supported when also passing a sa_column"
                 )
             if nullable is not Undefined:
                 raise RuntimeError(
                     "Passing nullable is not supported when also passing a sa_column"
                 )
             if foreign_key is not Undefined:
                 raise RuntimeError(
-                    "Passing foreign_key is not supported when "
-                    "also passing a sa_column"
+                    "Passing foreign_key is not supported when also passing a sa_column"
                 )
             if ondelete is not Undefined:
                 raise RuntimeError(

tests/test_column_property_deferred.py

@@ -0,0 +1,101 @@
+from typing import Optional
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import undefer
+from sqlmodel import Field, Session, SQLModel, deferred_column_property, select
+
+
+def test_deferred_column_property(clear_sqlmodel):
+    """Test deferred_column_property that returns fallback value instead of raising DetachedInstanceError"""
+
+    class DeferredModel(SQLModel, table=True):
+        __tablename__ = "deferred_model"
+
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        value: int = 0
+
+        @classmethod
+        def __declare_last__(cls):
+            # Use deferred_column_property instead of regular column_property
+            cls.computed_value = deferred_column_property(
+                cls.__table__.c.value * 2,
+                fallback_value=-1,
+                deferred=True,
+            )
+
+    engine = create_engine("sqlite://")
+    SQLModel.metadata.create_all(engine)
+
+    test_record = DeferredModel(name="Test", value=5)
+
+    with Session(engine) as session:
+        session.add(test_record)
+        session.commit()
+        session.refresh(test_record)
+        record_id = test_record.id
+
+    # Query without loading deferred property
+    with Session(engine) as session:
+        record = session.get(DeferredModel, record_id)
+        assert record is not None
+        assert record.name == "Test"
+        assert record.value == 5
+
+        # Close session to ensure deferred property cannot be loaded
+        session.close()
+
+        # Access deferred property - should return fallback value instead of raising error
+        computed = record.computed_value
+        assert computed == -1  # This is the key difference from regular column_property
+
+    # Test with explicit loading using undefer
+    with Session(engine) as session:
+        # Use undefer with class attribute instead of string
+        statement = (
+            select(DeferredModel)
+            .options(undefer(DeferredModel.computed_value))
+            .where(DeferredModel.id == record_id)
+        )
+        record = session.exec(statement).first()
+        assert record is not None
+
+        # Now the deferred property should be loaded and computed
+        computed = record.computed_value
+        assert computed == 10  # 5 * 2 = 10
+
+    # Test different fallback value
+    class CustomFallbackModel(SQLModel, table=True):
+        __tablename__ = "custom_fallback_model"
+
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        value: int = 0
+
+        @classmethod
+        def __declare_last__(cls):
+            cls.computed_value = deferred_column_property(
+                cls.__table__.c.value * 3,
+                fallback_value=999,  # Custom fallback value
+                deferred=True,
+            )
+
+    SQLModel.metadata.create_all(engine)
+
+    custom_record = CustomFallbackModel(name="Custom", value=7)
+
+    with Session(engine) as session:
+        session.add(custom_record)
+        session.commit()
+        session.refresh(custom_record)
+        custom_record_id = custom_record.id
+
+    # Test custom fallback value
+    with Session(engine) as session:
+        record = session.get(CustomFallbackModel, custom_record_id)
+        assert record is not None
+        session.close()
+
+        # Should return custom fallback value
+        computed = record.computed_value
+        assert computed == 999