MPT-22243 Support sqlalchemy hybrid properties

src/requela/builders/sqlalchemy.py

@@ -45,8 +45,11 @@ def get_field_type(self, field: str) -> type:
         parts = field.split(".")
         for part in parts[:-1]:
             model = getattr(model, part).property.mapper.class_
-        field_type = getattr(model, parts[-1]).property.columns[0].type.python_type
-        return field_type
+        attr = getattr(model, parts[-1])
+        prop = getattr(attr, "property", None)
+        if prop is not None and hasattr(prop, "columns"):
+            return prop.columns[0].type.python_type
+        return attr.type.python_type
 
     def apply_and(self, *conditions: ColumnExpressionArgument) -> ColumnElement:
         return and_(*conditions)
@@ -61,7 +64,7 @@ def apply_eq(
         self, prop: str, value: str | bool | date | datetime | int | float | None
     ) -> ColumnExpressionArgument:
         model_field = self.resolve_property(prop)
-        if isinstance(model_field.property, RelationshipProperty):
+        if isinstance(getattr(model_field, "property", None), RelationshipProperty):
             if value is not None:
                 raise ValueError("`eq` can be applied to relationship only to test for null.")
             return self.apply_eq_to_relationship(model_field.property)
@@ -82,7 +85,7 @@ def apply_ne(
         self, prop: str, value: str | bool | date | datetime | int | float | None
     ) -> ColumnExpressionArgument:
         model_field = self.resolve_property(prop)
-        if isinstance(model_field.property, RelationshipProperty):
+        if isinstance(getattr(model_field, "property", None), RelationshipProperty):
             if value is not None:
                 raise ValueError("`ne` can be applied to relationship only to test for null.")
             return self.apply_ne_to_relationship(model_field.property)

tests/sqlalchemy/models.py

@@ -4,6 +4,7 @@
 
 import sqlalchemy as sa
 from sqlalchemy import Boolean, Date, DateTime, Float, ForeignKey, Integer, String
+from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from sqlalchemy.types import Enum as SQLEnum
 
@@ -69,6 +70,19 @@ class User(Base):
     account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
     account: Mapped[Account] = relationship("Account", back_populates="users")
 
+    @hybrid_property
+    def is_inactive(self) -> bool:
+        return self.account.status == AccountStatus.INACTIVE
+
+    @is_inactive.inplace.expression
+    @classmethod
+    def _is_inactive_expr(cls):
+        return cls.account.has(Account.status == AccountStatus.INACTIVE)
+
+    @hybrid_property
+    def company_email(self) -> str:
+        return self.name + "." + self.role + "@requela.com"
+
 
 class ChargesFile(Base):
     __tablename__ = "invoices"

tests/sqlalchemy/rules.py

@@ -57,3 +57,6 @@ class UserRules(ModelRQLRules):
     status = FieldRule(
         source="account.status",
     )
+    is_inactive = FieldRule(allowed_operators=[Operator.EQ, Operator.NE])
+
+    company_email = FieldRule()

tests/sqlalchemy/test_rules.py

@@ -156,8 +156,10 @@ def test_get_documentation():
         "|account.tenant.name|eq, ilike, in, like, ne, out|yes|",
         "|account.tenant|eq, ne|no|",
         "|account|eq, ne|no|",
+        "|company_email|eq, ilike, in, like, ne, out|yes|",
         "|events.born.at|eq, gt, gte, lt, lte, ne|yes|",
         "|is_active|eq, ne|yes|",
+        "|is_inactive|eq, ne|yes|",
         "|name|eq, ilike, in, like, ne, out|yes|",
         "|role|in, out|yes|",
         "|status|eq, in, ne, out|yes|",
@@ -202,3 +204,14 @@ def test_valid_source_field():
         .join(account_alias, User.account)
     )
     assert_statements_equal(stmt, expected)
+
+
+def test_valid_hybrid_property():
+    user_rules = UserRules()
+    stmt = user_rules.build_query("eq(is_inactive,true)")
+    assert_statements_equal(stmt, select(User).filter(User.is_inactive.is_(True)))
+
+    stmt = user_rules.build_query("eq(company_email,john.user@requela.com)")
+    assert_statements_equal(
+        stmt, select(User).filter(User.company_email == "john.user@requela.com")
+    )