Merge pull request #54 from tomsch420/main

Fixed bug in getting relationships from parent alternatively mapped objects.

Author: tomsch420

src/krrood/ormatic/dao.py

@@ -12,6 +12,7 @@
 import sqlalchemy.orm
 from sqlalchemy import Column
 from sqlalchemy.orm import MANYTOONE, MANYTOMANY, ONETOMANY, RelationshipProperty
+from sqlalchemy.util import ReadOnlyProperties
 from typing_extensions import (
     Type,
     get_args,
@@ -23,6 +24,7 @@
     Optional,
     List,
     Iterable,
+    Tuple,
 )
 
 from ..utils import recursive_subclasses
@@ -527,17 +529,42 @@ def to_dao_if_subclass_of_alternative_mapping(
                 setattr(self, prop.key, getattr(obj, prop.key))
 
         # split relationships in relationships by parent and relationships by child
-        all_relationships = mapper.relationships
-        relationships_of_parent = parent_mapper.relationships
-        relationships_of_this_table = [
-            r for r in all_relationships if r not in relationships_of_parent
-        ]
+        relationships_of_parent, relationships_of_this_table = (
+            self.partition_parent_child_relationships(parent_mapper, mapper)
+        )
 
-        for relationship in relationships_of_parent:
-            setattr(self, relationship.key, getattr(parent_dao, relationship.key))
+        # get relationships from parent dao
+        self.get_relationships_from(parent_dao, relationships_of_parent, state)
 
+        # get relationships from the current table
         self.get_relationships_from(obj, relationships_of_this_table, state)
 
+    def partition_parent_child_relationships(
+        self, parent: sqlalchemy.orm.Mapper, child: sqlalchemy.orm.Mapper
+    ) -> Tuple[
+        List[RelationshipProperty[Any]],
+        List[RelationshipProperty[Any]],
+    ]:
+        """
+        Partition the relationships by parent-only and child-only relationships.
+
+        :param parent: The parent mapper to extract relationships from
+        :param child: The child mapper to extract relationships from
+        :return: A tuple of the relationships that are only in the parent and the relationships that are only in the child
+        """
+        all_relationships = child.relationships
+        relationships_of_parent = parent.relationships
+        relationship_names_of_parent = list(
+            map(lambda x: x.key, relationships_of_parent)
+        )
+
+        relationships_of_child = list(
+            filter(
+                lambda x: x.key not in relationship_names_of_parent, all_relationships
+            )
+        )
+        return relationships_of_parent, relationships_of_child
+
     def get_columns_from(self, obj: T, columns: Iterable[Column]) -> None:
         """
         Retrieves and assigns values from specified columns of a given object.

test/dataset/example_classes.py

@@ -510,6 +510,8 @@ def create_instance(cls, obj: T):
 class ParentAlternativelyMapped:
     base_attribute: float = 0
 
+    entities: List[Entity] = field(default_factory=list)
+
 
 @dataclass
 class ChildLevel1NormallyMapped(ParentAlternativelyMapped):
@@ -524,15 +526,19 @@ class ChildLevel2NormallyMapped(ChildLevel1NormallyMapped):
 @dataclass
 class ParentAlternativelyMappedMapping(AlternativeMapping[ParentAlternativelyMapped]):
     derived_attribute: str
+    entities: List[Entity]
 
     @classmethod
     def create_instance(cls, obj: T) -> Self:
-        return cls(str(obj.base_attribute))
+        return cls(str(obj.base_attribute), obj.entities)
 
     def create_from_dao(self) -> T:
         raise NotImplementedError
 
 
+# %% Function like classes for testing
+
+
 @dataclass
 class CallableWrapper:
     func: FunctionType
@@ -560,6 +566,9 @@ class UUIDWrapper:
     other_identifications: List[uuid.UUID] = field(default_factory=list)
 
 
+# %% Test JSON serialization in ORM classes
+
+
 @dataclass
 class JSONSerializableClass(SubclassJSONSerializer):
     a: float = 0.0

test/dataset/ormatic_interface.py

@@ -44,6 +44,15 @@ class Base(DeclarativeBase):
 
 
 # Association tables for many-to-many relationships
+parentalternativelymappedmappingdao_entities_association = Table(
+    "parentalternativelymappedmappingdao_entities_association",
+    Base.metadata,
+    Column(
+        "parentalternativelymappedmappingdao_id",
+        ForeignKey("ParentAlternativelyMappedMappingDAO.database_id"),
+    ),
+    Column("customentitydao_id", ForeignKey("CustomEntityDAO.database_id")),
+)
 alternativemappingaggregatordao_entities1_association = Table(
     "alternativemappingaggregatordao_entities1_association",
     Base.metadata,
@@ -333,6 +342,12 @@ class ParentAlternativelyMappedMappingDAO(
         String(255), nullable=False, use_existing_column=True
     )
 
+    entities: Mapped[typing.List[CustomEntityDAO]] = relationship(
+        "CustomEntityDAO",
+        secondary="parentalternativelymappedmappingdao_entities_association",
+        cascade="save-update, merge",
+    )
+
     __mapper_args__ = {
         "polymorphic_on": "polymorphic_type",
         "polymorphic_identity": "ParentAlternativelyMappedMappingDAO",

test/test_ormatic/test_interface.py

@@ -475,10 +475,18 @@ def test_inheritance_mapper_args(session, database):
 
 
 def test_to_dao_alternatively_mapped_parent(session, database):
-    ch2 = ChildLevel2NormallyMapped(1, 2, 3)
+    ch2 = ChildLevel2NormallyMapped(1, [Entity("a")], 2, 3)
     ch2_dao = to_dao(ch2)
 
-    assert ch2_dao == ChildLevel2NormallyMappedDAO("1", 2, 3)
+    assert isinstance(ch2_dao.entities[0], CustomEntityDAO)
+    assert ch2_dao.entities == [CustomEntityDAO(overwritten_name="a")]
+
+    assert ch2_dao == ChildLevel2NormallyMappedDAO(
+        derived_attribute="1",
+        entities=[CustomEntityDAO(overwritten_name="a")],
+        level_one_attribute=2,
+        level_two_attribute=3,
+    )
 
 
 def test_callable_alternative_mapping():