feat(litestar): use property in SQLAlchemyDTO with MappedAsDataclass (#447)

Allow better compatibility with `MappedAsDataclass` and the `SQLALchemyDTO`

Author: cofin

advanced_alchemy/extensions/litestar/dto.py

@@ -1,7 +1,10 @@
+# ruff: noqa: C901
+import inspect as stdlib_inspect
+import logging
 from collections.abc import Collection, Generator
 from collections.abc import Set as AbstractSet
 from dataclasses import asdict, dataclass, field, replace
-from functools import singledispatchmethod
+from functools import cached_property, singledispatchmethod
 from typing import (
     Any,
     ClassVar,
@@ -44,6 +47,8 @@
 
 __all__ = ("SQLAlchemyDTO",)
 
+logger = logging.getLogger(__name__)
+
 T = TypeVar("T", bound="Union[DeclarativeBase, Collection[DeclarativeBase]]")
 
 ElementType: TypeAlias = Union[
@@ -231,7 +236,9 @@ def _(
                     default=Empty,
                 ),
                 default_factory=None,
-                dto_field=orm_descriptor.info.get(DTO_FIELD_META_KEY, DTOField(mark=Mark.READ_ONLY)),
+                dto_field=orm_descriptor.info.get(
+                    DTO_FIELD_META_KEY, DTOField(mark=Mark.READ_ONLY)
+                ),  # Mark as read-only
                 model_name=model_name,
             ),
         ]
@@ -260,7 +267,9 @@ def _(
                     default=Empty,
                 ),
                 default_factory=None,
-                dto_field=orm_descriptor.info.get(DTO_FIELD_META_KEY, DTOField(mark=Mark.READ_ONLY)),
+                dto_field=orm_descriptor.info.get(
+                    DTO_FIELD_META_KEY, DTOField(mark=Mark.READ_ONLY)
+                ),  # Mark as read-only
                 model_name=model_name,
             ),
         ]
@@ -275,13 +284,61 @@ def _(
                         default=Empty,
                     ),
                     default_factory=None,
-                    dto_field=orm_descriptor.info.get(DTO_FIELD_META_KEY, DTOField(mark=Mark.WRITE_ONLY)),
+                    dto_field=orm_descriptor.info.get(
+                        DTO_FIELD_META_KEY, DTOField(mark=Mark.WRITE_ONLY)
+                    ),  # Mark as read-only
                     model_name=model_name,
                 ),
             )
 
         return field_defs
 
+    @classmethod
+    def get_property_fields(cls, model_type: "type[DeclarativeBase]") -> "dict[str, FieldDefinition]":
+        """Get fields defined as @property or @cached_property on the model.
+
+        Uses inspect.getmembers() to detect properties from the model class and mixins.
+        Properties are marked read-only; setter support is not implemented.
+
+        Args:
+            model_type: The SQLAlchemy model type to extract properties from.
+
+        Returns:
+            A dictionary mapping property names to their field definitions.
+        """
+        namespace = cls.get_model_namespace(model_type)
+        sqla_internal_properties = {"awaitable_attrs", "registry", "metadata"}
+
+        properties: dict[str, FieldDefinition] = {}
+        for name, member in stdlib_inspect.getmembers(
+            model_type, predicate=lambda x: isinstance(x, (property, cached_property))
+        ):
+            if name in sqla_internal_properties:
+                continue
+
+            if isinstance(member, cached_property):
+                func = member.func
+            elif isinstance(member, property):
+                if member.fget is None:
+                    continue
+                func = member.fget
+            else:
+                continue
+
+            try:
+                sig = ParsedSignature.from_fn(func, namespace)
+                properties[name] = replace(sig.return_type, name=name)
+            except (AttributeError, TypeError, ValueError) as e:
+                logger.debug(
+                    "could not parse type hint for property %s.%s: %s, using Any type",
+                    model_type.__name__,
+                    name,
+                    e,
+                )
+                properties[name] = FieldDefinition.from_annotation(Any, name=name)
+
+        return properties
+
     @classmethod
     def generate_field_definitions(cls, model_type: type[DeclarativeBase]) -> Generator[DTOFieldDefinition, None, None]:
         """Generate DTO field definitions from a SQLAlchemy model.
@@ -315,6 +372,9 @@ def generate_field_definitions(cls, model_type: type[DeclarativeBase]) -> Genera
                     skipped_descriptors.add(attr.name)
                 elif isinstance(attr, str):
                     skipped_descriptors.add(attr)
+
+        yielded_sqla_keys: set[str] = set()  # Keep track of keys yielded by SQLAlchemy logic
+
         for key, orm_descriptor in mapper.all_orm_descriptors.items():
             if is_hybrid_property := isinstance(orm_descriptor, hybrid_property):
                 if orm_descriptor in seen_hybrid_descriptors:
@@ -328,7 +388,12 @@ def generate_field_definitions(cls, model_type: type[DeclarativeBase]) -> Genera
             should_skip_descriptor = False
             dto_field: Optional[DTOField] = None
             if hasattr(orm_descriptor, "property"):  # pyright: ignore[reportUnknownArgumentType]
-                dto_field = orm_descriptor.property.info.get(DTO_FIELD_META_KEY)  # pyright: ignore
+                # Access info safely, checking if property exists first
+                prop = getattr(orm_descriptor, "property", None)  # pyright: ignore[reportUnknownArgumentType]
+                if prop and hasattr(prop, "info"):
+                    dto_field = prop.info.get(DTO_FIELD_META_KEY)
+                elif hasattr(orm_descriptor, "info"):  # pyright: ignore[reportUnknownArgumentType]
+                    dto_field = orm_descriptor.info.get(DTO_FIELD_META_KEY)  # pyright: ignore[reportUnknownArgumentType,reportUnknownMemberType,reportAttributeAccessIssue,reportUnknownVariableType]
 
             # Case 1
             is_field_marked_not_private = dto_field and dto_field.mark is not Mark.PRIVATE  # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType]
@@ -353,13 +418,33 @@ def generate_field_definitions(cls, model_type: type[DeclarativeBase]) -> Genera
             if should_skip_descriptor:
                 continue
 
-            yield from cls.handle_orm_descriptor(
+            # Yield definitions from SQLAlchemy descriptor handling
+            definitions = cls.handle_orm_descriptor(
                 orm_descriptor.extension_type,
                 key,
                 orm_descriptor,
                 model_type_hints,
                 model_name,
             )
+            for definition in definitions:
+                yielded_sqla_keys.add(definition.name)  # Track yielded key
+                yield definition
+
+        property_fields = cls.get_property_fields(model_type)
+        for key, property_field_definition in property_fields.items():
+            if key.startswith("_") or key in yielded_sqla_keys:
+                continue
+
+            yield DTOFieldDefinition.from_field_definition(
+                field_definition=replace(
+                    property_field_definition,
+                    name=key,
+                    default=Empty,
+                ),
+                model_name=model_name,
+                default_factory=None,
+                dto_field=DTOField(mark=Mark.READ_ONLY),
+            )
 
     @classmethod
     def detect_nested_field(cls, field_definition: FieldDefinition) -> bool:

docs/usage/frameworks/litestar.rst

@@ -70,6 +70,47 @@ Define your SQLAlchemy models using Advanced Alchemy's enhanced base classes:
         author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
         author: Mapped[AuthorModel] = relationship(lazy="joined", innerjoin=True, viewonly=True)
 
+Using Properties with DTOs
+---------------------------
+
+SQLAlchemyDTO includes Python ``@property`` and ``@functools.cached_property`` decorated methods as read-only fields.
+
+.. code-block:: python
+
+    from functools import cached_property
+    from sqlalchemy.orm import Mapped, mapped_column, MappedAsDataclass
+    from advanced_alchemy.extensions.litestar import base, SQLAlchemyDTO
+
+    class UserModel(base.UUIDAuditBase, MappedAsDataclass):
+        __tablename__ = "user"
+
+        first_name: Mapped[str]
+        last_name: Mapped[str]
+
+        @property
+        def full_name(self) -> str:
+            return f"{self.first_name} {self.last_name}"
+
+        @cached_property
+        def name_length(self) -> int:
+            return len(self.full_name)
+
+    # DTO includes: id, created_at, updated_at, first_name, last_name,
+    # full_name (read-only), name_length (read-only)
+    UserDTO = SQLAlchemyDTO[UserModel]
+
+Property handling characteristics:
+
+- Detected from model class and mixins
+- Marked as ``READ_ONLY`` (cannot be set via DTO)
+- Type inferred from return type annotations
+- Private properties (starting with ``_``) excluded
+- Skipped if already handled by SQLAlchemy descriptors (e.g., ``hybrid_property``)
+
+.. note::
+
+    Properties with setters (``@property.setter``) are marked ``READ_ONLY``. Setter support is not implemented.
+
 Pydantic Schemas
 ----------------
 

tests/unit/test_extensions/test_litestar/test_dto_integration.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from functools import cached_property
 from types import ModuleType
 from typing import Annotated, Any, Callable, Optional
 from uuid import UUID
@@ -230,6 +231,124 @@ async def get_handler() -> ModelWithFunc:
         assert response_callback
 
 
+def test_dto_includes_simple_property() -> None:
+    """Test that @property decorated methods appear in DTO as read-only fields."""
+
+    class ModelWithProperty(Base):
+        __tablename__ = "model_with_property"
+
+        first_name: Mapped[str] = mapped_column()
+        last_name: Mapped[str] = mapped_column()
+
+        @property
+        def full_name(self) -> str:
+            return f"{self.first_name} {self.last_name}"
+
+    config = SQLAlchemyDTOConfig()
+    dto = SQLAlchemyDTO[Annotated[ModelWithProperty, config]]
+    field_defs = list(dto.generate_field_definitions(ModelWithProperty))
+    field_names = {f.name for f in field_defs}
+
+    assert "full_name" in field_names
+
+    full_name_field = next(f for f in field_defs if f.name == "full_name")
+    assert full_name_field.dto_field.mark == Mark.READ_ONLY
+
+
+def test_dto_includes_cached_property() -> None:
+    """Test that @cached_property decorated methods appear in DTO as read-only fields."""
+
+    class ModelWithCachedProperty(Base):
+        __tablename__ = "model_with_cached_property"
+
+        value: Mapped[int] = mapped_column()
+
+        @cached_property
+        def expensive_calculation(self) -> int:
+            return self.value * 2
+
+    config = SQLAlchemyDTOConfig()
+    dto = SQLAlchemyDTO[Annotated[ModelWithCachedProperty, config]]
+    field_defs = list(dto.generate_field_definitions(ModelWithCachedProperty))
+    field_names = {f.name for f in field_defs}
+
+    assert "expensive_calculation" in field_names
+
+    field = next(f for f in field_defs if f.name == "expensive_calculation")
+    assert field.dto_field.mark == Mark.READ_ONLY
+
+
+def test_dto_property_with_setter_is_read_only() -> None:
+    """Test that properties with setters are marked READ_ONLY (setter support not implemented)."""
+
+    class ModelWithPropertySetter(Base):
+        __tablename__ = "model_with_property_setter"
+
+        _internal_value: Mapped[int] = mapped_column(default=0)
+
+        @property
+        def value(self) -> int:
+            return self._internal_value
+
+        @value.setter
+        def value(self, new_value: int) -> None:
+            self._internal_value = new_value
+
+    config = SQLAlchemyDTOConfig()
+    dto = SQLAlchemyDTO[Annotated[ModelWithPropertySetter, config]]
+    field_defs = list(dto.generate_field_definitions(ModelWithPropertySetter))
+    field_names = {f.name for f in field_defs}
+
+    assert "value" in field_names
+
+    field = next(f for f in field_defs if f.name == "value")
+    assert field.dto_field.mark == Mark.READ_ONLY
+
+
+def test_dto_skips_private_properties() -> None:
+    """Test that properties starting with _ are excluded from DTO."""
+
+    class ModelWithPrivateProperty(Base):
+        __tablename__ = "model_with_private_property"
+
+        @property
+        def public_prop(self) -> str:
+            return "public"
+
+        @property
+        def _private_prop(self) -> str:
+            return "private"
+
+    config = SQLAlchemyDTOConfig()
+    dto = SQLAlchemyDTO[Annotated[ModelWithPrivateProperty, config]]
+    field_defs = list(dto.generate_field_definitions(ModelWithPrivateProperty))
+    field_names = {f.name for f in field_defs}
+
+    assert "public_prop" in field_names
+    assert "_private_prop" not in field_names
+
+
+def test_dto_handles_untyped_property() -> None:
+    """Test that properties without type hints are included with Any type."""
+
+    class ModelWithUntypedProperty(Base):
+        __tablename__ = "model_with_untyped_property"
+
+        @property
+        def untyped_prop(self):  # type: ignore[no-untyped-def]
+            return "value"
+
+    config = SQLAlchemyDTOConfig()
+    dto = SQLAlchemyDTO[Annotated[ModelWithUntypedProperty, config]]
+    field_defs = list(dto.generate_field_definitions(ModelWithUntypedProperty))
+    field_names = {f.name for f in field_defs}
+
+    assert "untyped_prop" in field_names
+
+    field = next(f for f in field_defs if f.name == "untyped_prop")
+    assert field is not None
+
+
 def test_dto_with_association_proxy(create_module: Callable[[str], ModuleType]) -> None:
     module = create_module(
         """