fix: deduplicate relationship joins when searchable_fields and facet_fields reference the same model (#217)

src/fastapi_toolsets/crud/factory.py

@@ -116,8 +116,12 @@ def _apply_joins(q: Any, joins: JoinType | None, outer_join: bool) -> Any:
 
 def _apply_search_joins(q: Any, search_joins: list[Any]) -> Any:
     """Apply relationship-based outer joins (from search/filter_by) to a query."""
+    seen: set[str] = set()
     for join_rel in search_joins:
-        q = q.outerjoin(join_rel)
+        key = str(join_rel)
+        if key not in seen:
+            seen.add(key)
+            q = q.outerjoin(join_rel)
     return q
 
 

src/fastapi_toolsets/crud/search.py

@@ -242,13 +242,19 @@ async def build_facets(
         else:
             q = select(column).select_from(model).distinct()
 
-        # Apply base joins (already done on main query, but needed here independently)
+        # Apply base joins (deduplicated) — needed here independently
+        seen_joins: set[str] = set()
         for rel in base_joins or []:
-            q = q.outerjoin(rel)
+            rel_key = str(rel)
+            if rel_key not in seen_joins:
+                seen_joins.add(rel_key)
+                q = q.outerjoin(rel)
 
-        # Add any extra joins required by this facet field that aren't already in base_joins
+        # Add any extra joins required by this facet field that aren't already applied
         for rel in rels:
-            if str(rel) not in existing_join_keys:
+            rel_key = str(rel)
+            if rel_key not in existing_join_keys and rel_key not in seen_joins:
+                seen_joins.add(rel_key)
                 q = q.outerjoin(rel)
 
         if base_filters:

tests/test_crud_search.py

@@ -697,6 +697,67 @@ class TestFacetsRelationship:
         assert result.filter_attributes is not None
         assert result.filter_attributes["role__name"] == ["admin"]
 
+    @pytest.mark.anyio
+    async def test_relationship_search_and_filter_by_same_join(
+        self, db_session: AsyncSession
+    ):
+        """Search + filter_by on the same relationship must not duplicate the JOIN."""
+        UserSearchFacetCrud = CrudFactory(
+            User,
+            searchable_fields=[(User.role, Role.name)],
+            facet_fields=[(User.role, Role.name)],
+        )
+
+        admin = await RoleCrud.create(db_session, RoleCreate(name="admin"))
+        editor = await RoleCrud.create(db_session, RoleCreate(name="editor"))
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="alice", email="a@test.com", role_id=admin.id),
+        )
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="bob", email="b@test.com", role_id=editor.id),
+        )
+
+        # Search by role name AND filter by role name — both need the same join
+        result = await UserSearchFacetCrud.offset_paginate(
+            db_session,
+            search="admin",
+            filter_by={"role__name": "admin"},
+            schema=UserRead,
+        )
+
+        assert len(result.data) == 1
+        assert result.data[0].username == "alice"
+        assert result.filter_attributes is not None
+        assert result.filter_attributes["role__name"] == ["admin"]
+
+    @pytest.mark.anyio
+    async def test_cursor_paginate_duplicate_join(self, db_session: AsyncSession):
+        """cursor_paginate with overlapping search + facet joins must not fail."""
+        UserSearchFacetCursorCrud = CrudFactory(
+            User,
+            searchable_fields=[(User.role, Role.name)],
+            facet_fields=[(User.role, Role.name)],
+            cursor_column=User.id,
+        )
+
+        admin = await RoleCrud.create(db_session, RoleCreate(name="admin"))
+        await UserCrud.create(
+            db_session,
+            UserCreate(username="alice", email="a@test.com", role_id=admin.id),
+        )
+
+        result = await UserSearchFacetCursorCrud.cursor_paginate(
+            db_session,
+            search="admin",
+            filter_by={"role__name": "admin"},
+            schema=UserRead,
+        )
+
+        assert len(result.data) == 1
+        assert result.data[0].username == "alice"
+
 
 class TestFilterBy:
     """Tests for the filter_by parameter on offset_paginate and cursor_paginate."""