fix overload and implementation of get_joined

Author: igorbenav

fastcrud/crud/fast_crud.py

@@ -1401,10 +1401,91 @@ async def get_multi(
 
         return response
 
+    @overload
+    async def get_joined(
+        self,
+        db: AsyncSession,
+        *,
+        schema_to_select: type[SelectSchemaType],
+        return_as_model: Literal[True],
+        join_model: Optional[ModelType] = None,
+        join_on: Optional[Union[Join, BinaryExpression]] = None,
+        join_prefix: Optional[str] = None,
+        join_schema_to_select: Optional[type[SelectSchemaType]] = None,
+        join_type: str = "left",
+        alias: Optional[AliasedClass] = None,
+        join_filters: Optional[dict] = None,
+        joins_config: Optional[list[JoinConfig]] = None,
+        nest_joins: bool = False,
+        relationship_type: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Optional[SelectSchemaType]: ...
+
+    @overload
+    async def get_joined(
+        self,
+        db: AsyncSession,
+        *,
+        schema_to_select: None = None,
+        return_as_model: Literal[False] = False,
+        join_model: Optional[ModelType] = None,
+        join_on: Optional[Union[Join, BinaryExpression]] = None,
+        join_prefix: Optional[str] = None,
+        join_schema_to_select: Optional[type[SelectSchemaType]] = None,
+        join_type: str = "left",
+        alias: Optional[AliasedClass] = None,
+        join_filters: Optional[dict] = None,
+        joins_config: Optional[list[JoinConfig]] = None,
+        nest_joins: bool = False,
+        relationship_type: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Optional[dict[str, Any]]: ...
+
+    @overload
+    async def get_joined(
+        self,
+        db: AsyncSession,
+        *,
+        schema_to_select: type[SelectSchemaType],
+        return_as_model: Literal[False] = False,
+        join_model: Optional[ModelType] = None,
+        join_on: Optional[Union[Join, BinaryExpression]] = None,
+        join_prefix: Optional[str] = None,
+        join_schema_to_select: Optional[type[SelectSchemaType]] = None,
+        join_type: str = "left",
+        alias: Optional[AliasedClass] = None,
+        join_filters: Optional[dict] = None,
+        joins_config: Optional[list[JoinConfig]] = None,
+        nest_joins: bool = False,
+        relationship_type: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Optional[dict[str, Any]]: ...
+
+    @overload
+    async def get_joined(
+        self,
+        db: AsyncSession,
+        *,
+        schema_to_select: Optional[type[SelectSchemaType]] = None,
+        return_as_model: bool = False,
+        join_model: Optional[ModelType] = None,
+        join_on: Optional[Union[Join, BinaryExpression]] = None,
+        join_prefix: Optional[str] = None,
+        join_schema_to_select: Optional[type[SelectSchemaType]] = None,
+        join_type: str = "left",
+        alias: Optional[AliasedClass] = None,
+        join_filters: Optional[dict] = None,
+        joins_config: Optional[list[JoinConfig]] = None,
+        nest_joins: bool = False,
+        relationship_type: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Optional[Union[dict[str, Any], SelectSchemaType]]: ...
+
     async def get_joined(
         self,
         db: AsyncSession,
         schema_to_select: Optional[type[SelectSchemaType]] = None,
+        return_as_model: bool = False,
         join_model: Optional[ModelType] = None,
         join_on: Optional[Union[Join, BinaryExpression]] = None,
         join_prefix: Optional[str] = None,
@@ -1416,7 +1497,7 @@ async def get_joined(
         nest_joins: bool = False,
         relationship_type: Optional[str] = None,
         **kwargs: Any,
-    ) -> Optional[dict[str, Any]]:
+    ) -> Optional[Union[dict[str, Any], SelectSchemaType]]:
         """
         Fetches a single record with one or multiple joins on other models. If `join_on` is not provided, the method attempts
         to automatically detect the join condition using foreign key relationships. For multiple joins, use `joins_config` to
@@ -1427,6 +1508,7 @@ async def get_joined(
         Args:
             db: The SQLAlchemy async session.
             schema_to_select: Pydantic schema for selecting specific columns from the primary model. Required if `return_as_model` is True.
+            return_as_model: If `True`, returns data as a Pydantic model instance based on `schema_to_select`. Defaults to `False`.
             join_model: The model to join with.
             join_on: SQLAlchemy Join object for specifying the `ON` clause of the join. If `None`, the join condition is auto-detected based on foreign keys.
             join_prefix: Optional prefix to be added to all columns of the joined model. If `None`, no prefix is added.
@@ -1440,7 +1522,10 @@ async def get_joined(
             **kwargs: Filters to apply to the primary model query, supporting advanced comparison operators for refined searching.
 
         Returns:
-            A dictionary representing the joined record, or `None` if no record matches the criteria.
+            A dictionary or Pydantic model instance representing the joined record, or `None` if no record matches the criteria:
+
+            - When `return_as_model=True` and `schema_to_select` is provided: `Optional[SelectSchemaType]`
+            - When `return_as_model=False`: `Optional[Dict[str, Any]]`
 
         Raises:
             ValueError: If both single join parameters and `joins_config` are used simultaneously.
@@ -1708,7 +1793,19 @@ async def get_joined(
             else:
                 data_list = []
 
-        return process_joined_data(data_list, join_definitions, nest_joins, self.model)
+        processed_data = process_joined_data(
+            data_list, join_definitions, nest_joins, self.model
+        )
+
+        if processed_data is None or not return_as_model:
+            return processed_data
+
+        if not schema_to_select:
+            raise ValueError(
+                "schema_to_select must be provided when return_as_model is True."
+            )
+
+        return schema_to_select(**processed_data)
 
     @overload
     async def get_multi_joined(

tests/sqlalchemy/crud/test_get_joined.py

@@ -1,4 +1,6 @@
 import pytest
+from typing import Optional
+from pydantic import BaseModel
 from sqlalchemy import and_
 from fastcrud import FastCRUD, JoinConfig, aliased
 from ...sqlalchemy.conftest import (
@@ -24,6 +26,22 @@
 )
 
 
+# Schema that includes joined fields for testing
+class JoinedTestTier(BaseModel):
+    name: str
+    tier_id: int
+    tier_name: str
+
+
+# Flattened schema for task with joined fields
+class TaskWithJoinedData(BaseModel):
+    id: int
+    name: str
+    description: Optional[str] = None
+    assignee_name: Optional[str] = None
+    department_name: Optional[str] = None
+
+
 @pytest.mark.asyncio
 async def test_get_joined_basic(async_session, test_data, test_data_tier):
     for tier_item in test_data_tier:
@@ -713,3 +731,156 @@ async def test_get_joined_nested_data_none_dict(async_session):
     assert task3_result["client"] is None, "Task 3 should have no client."
     assert task3_result["department"] is None, "Task 3 should have no department."
     assert task3_result["assignee"] is None, "Task 3 should have no assignee."
+
+
+@pytest.mark.asyncio
+async def test_get_joined_return_as_model_true(
+    async_session, test_data, test_data_tier
+):
+    """Test get_joined with return_as_model=True returns Pydantic model instance."""
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    await async_session.commit()
+
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    crud = FastCRUD(ModelTest)
+    result = await crud.get_joined(
+        db=async_session,
+        join_model=TierModel,
+        join_prefix="tier_",
+        schema_to_select=JoinedTestTier,
+        join_schema_to_select=TierSchemaTest,
+        return_as_model=True,
+    )
+
+    assert result is not None
+    assert isinstance(
+        result, JoinedTestTier
+    ), "Result should be a JoinedTestTier Pydantic model instance"
+    assert hasattr(result, "name"), "Result should have name attribute"
+    assert hasattr(result, "tier_name"), "Result should have tier_name attribute"
+    assert result.tier_name is not None, "tier_name should have a value"
+
+
+@pytest.mark.asyncio
+async def test_get_joined_return_as_model_false(
+    async_session, test_data, test_data_tier
+):
+    """Test get_joined with return_as_model=False returns dict (default behavior)."""
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    await async_session.commit()
+
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    crud = FastCRUD(ModelTest)
+    result = await crud.get_joined(
+        db=async_session,
+        join_model=TierModel,
+        join_prefix="tier_",
+        schema_to_select=ReadSchemaTest,
+        join_schema_to_select=TierSchemaTest,
+        return_as_model=False,
+    )
+
+    assert result is not None
+    assert isinstance(result, dict), "Result should be a dictionary"
+    assert "name" in result
+    assert "tier_name" in result
+
+
+@pytest.mark.asyncio
+async def test_get_joined_return_as_model_without_schema_raises_error(
+    async_session, test_data, test_data_tier
+):
+    """Test get_joined with return_as_model=True but no schema raises ValueError."""
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    await async_session.commit()
+
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    crud = FastCRUD(ModelTest)
+    with pytest.raises(ValueError) as exc_info:
+        await crud.get_joined(
+            db=async_session,
+            join_model=TierModel,
+            join_prefix="tier_",
+            return_as_model=True,  # Missing schema_to_select
+        )
+
+    assert "schema_to_select must be provided when return_as_model is True" in str(
+        exc_info.value
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_joined_return_as_model_with_joins_config(async_session):
+    """Test get_joined with return_as_model=True using joins_config."""
+    # Create test data
+    department = Department(name="Engineering")
+    async_session.add(department)
+    await async_session.flush()
+
+    user = User(
+        name="John Doe",
+        username="john",
+        email="[email protected]",
+        department_id=department.id,
+    )
+    async_session.add(user)
+    await async_session.flush()
+
+    task = Task(
+        name="Test Task",
+        description="Test Task",
+        assignee_id=user.id,
+        department_id=department.id,
+    )
+    async_session.add(task)
+    await async_session.commit()
+
+    # Test with joins_config and return_as_model=True
+    task_crud = FastCRUD(Task)
+    joins_config = [
+        JoinConfig(
+            model=User,
+            join_on=Task.assignee_id == User.id,
+            join_prefix="assignee_",
+            schema_to_select=UserReadSub,
+            join_type="left",
+        ),
+        JoinConfig(
+            model=Department,
+            join_on=Task.department_id == Department.id,
+            join_prefix="department_",
+            schema_to_select=DepartmentRead,
+            join_type="left",
+        ),
+    ]
+
+    result = await task_crud.get_joined(
+        db=async_session,
+        id=task.id,
+        schema_to_select=TaskWithJoinedData,
+        joins_config=joins_config,
+        return_as_model=True,
+    )
+
+    assert result is not None
+    assert isinstance(
+        result, TaskWithJoinedData
+    ), "Result should be a TaskWithJoinedData Pydantic model instance"
+    assert hasattr(result, "description"), "Result should have description attribute"
+    assert result.description == "Test Task"
+    assert hasattr(
+        result, "assignee_name"
+    ), "Result should have assignee_name attribute"
+    assert result.assignee_name == "John Doe"