Repository: igorbenav/fastcrud (python)

Issue:
Preventing Duplicate Table Names in SQL Queries Using SQLAlchemy
Thank you for the recent update! We appreciate the enhancements and improvements made. It's not critical, but I think it's worth discussing.

**Describe the bug or question**
The problem lies in encountering an error due to duplicate table names when writing SQL queries using SQLAlchemy. This occurs when there's ambiguity in the query because the same table name is used multiple times without explicitly specifying aliases. As a result, SQLAlchemy fails to process the query correctly, leading to a query execution error.

**To Reproduce**
```python
booking_join_config = [
    JoinConfig(
        model=models.UserModel,
        join_on=models.Booking.owner_id == models.UserModel.id,
        join_prefix="owner_",
        schema_to_select=schemas.UserBase,
        join_type="inner",
    ),
    JoinConfig(
        model=models.UserModel,
        join_on=models.Booking.user_id == models.UserModel.id,
        join_prefix="user_",
        schema_to_select=schemas.UserBase,
        join_type="inner",
    ),
]
```

**Description**
Output: table name "users" specified more than once

```SQL
FROM bookings 
JOIN users ON bookings.owner_id = users.id 
JOIN users ON bookings.user_id = users.id 
```

**Additional context**
```SQL
FROM bookings 
JOIN users AS owner_users ON bookings.owner_id = owner_users.id 
JOIN users AS user_users ON bookings.user_id = user_users.id 
```


Pull request:
Using Aliases
# Using aliases
## Added
- Now get_joined and get_multi_joined can be used with aliases, making it possible to join the same model multiple times.

## Detailed
___

In complex query scenarios, particularly when you need to join a table to itself or perform multiple joins on the same table for different purposes, aliasing becomes crucial. Aliasing allows you to refer to the same table in different contexts with unique identifiers, avoiding conflicts and ambiguity in your queries.

For both `get_joined` and `get_multi_joined` methods, when you need to join the same model multiple times, you can utilize the `alias` parameter within your `JoinConfig` to differentiate between the joins. This parameter expects an instance of `AliasedClass`, which can be created using the `aliased` function from SQLAlchemy (also in fastcrud for convenience).

#### Example: Joining the Same Model Multiple Times

Consider a task management application where tasks have both an owner and an assigned user, represented by the same `UserModel`. To fetch tasks with details of both users, we use aliases to join the `UserModel` twice, distinguishing between owners and assigned users.

Let's start by creating the aliases and passing them to the join configuration. Don't forget to use the alias for `join_on`:

```python hl_lines="4-5 11 15 19 23" title="Join Configurations with Aliases"
from fastcrud import FastCRUD, JoinConfig, aliased

# Create aliases for UserModel to distinguish between the owner and the assigned user
owner_alias = aliased(UserModel, name="owner")
assigned_user_alias = aliased(UserModel, name="assigned_user")

# Configure joins with aliases
joins_config = [
    JoinConfig(
        model=UserModel,
        join_on=Task.owner_id == owner_alias.id,
        join_prefix="owner_",
        schema_to_select=UserSchema,
        join_type="inner",
        alias=owner_alias  # Pass the aliased class instance
    ),
    JoinConfig(
        model=UserModel,
        join_on=Task.assigned_user_id == assigned_user_alias.id,
        join_prefix="assigned_",
        schema_to_select=UserSchema,
        join_type="inner",
        alias=assigned_user_alias  # Pass the aliased class instance
    )
]

# Initialize your FastCRUD instance for TaskModel
task_crud = FastCRUD(TaskModel)

# Fetch tasks with joined user details
tasks = await task_crud.get_multi_joined(
    db=session,
    schema_to_select=TaskSchema,
    joins_config=joins_config,
    offset=0,
    limit=10
)
```

Then just pass this joins_config to `get_multi_joined`:

```python hl_lines="17" title="Passing joins_config to get_multi_joined"
from fastcrud import FastCRUD, JoinConfig, aliased

...

# Configure joins with aliases
joins_config = [
    ...
]

# Initialize your FastCRUD instance for TaskModel
task_crud = FastCRUD(TaskModel)

# Fetch tasks with joined user details
tasks = await task_crud.get_multi_joined(
    db=session,
    schema_to_select=TaskSchema,
    joins_config=joins_config,
    offset=0,
    limit=10
)
```

In this example, `owner_alias` and `assigned_user_alias` are created from `UserModel` to distinguish between the task's owner and the assigned user within the task management system. By using aliases, you can join the same model multiple times for different purposes in your queries, enhancing expressiveness and eliminating ambiguity.

Closes #27 

Patch:
diff --git a/docs/advanced/crud.md b/docs/advanced/crud.md
index 4404ac5..3dfef13 100644
--- a/docs/advanced/crud.md
+++ b/docs/advanced/crud.md
@@ -133,6 +133,85 @@ In this example, users are joined with the `Tier` and `Department` models. The `
 
     If both single join parameters and `joins_config` are used simultaneously, an error will be raised.
 
+### Using aliases
+
+In complex query scenarios, particularly when you need to join a table to itself or perform multiple joins on the same table for different purposes, aliasing becomes crucial. Aliasing allows you to refer to the same table in different contexts with unique identifiers, avoiding conflicts and ambiguity in your queries.
+
+For both `get_joined` and `get_multi_joined` methods, when you need to join the same model multiple times, you can utilize the `alias` parameter within your `JoinConfig` to differentiate between the joins. This parameter expects an instance of `AliasedClass`, which can be created using the `aliased` function from SQLAlchemy (also in fastcrud for convenience).
+
+#### Example: Joining the Same Model Multiple Times
+
+Consider a task management application where tasks have both an owner and an assigned user, represented by the same `UserModel`. To fetch tasks with details of both users, we use aliases to join the `UserModel` twice, distinguishing between owners and assigned users.
+
+Let's start by creating the aliases and passing them to the join configuration. Don't forget to use the alias for `join_on`:
+
+```python hl_lines="4-5 11 15 19 23" title="Join Configurations with Aliases"
+from fastcrud import FastCRUD, JoinConfig, aliased
+
+# Create aliases for UserModel to distinguish between the owner and the assigned user
+owner_alias = aliased(UserModel, name="owner")
+assigned_user_alias = aliased(UserModel, name="assigned_user")
+
+# Configure joins with aliases
+joins_config = [
+    JoinConfig(
+        model=UserModel,
+        join_on=Task.owner_id == owner_alias.id,
+        join_prefix="owner_",
+        schema_to_select=UserSchema,
+        join_type="inner",
+        alias=owner_alias  # Pass the aliased class instance
+    ),
+    JoinConfig(
+        model=UserModel,
+        join_on=Task.assigned_user_id == assigned_user_alias.id,
+        join_prefix="assigned_",
+        schema_to_select=UserSchema,
+        join_type="inner",
+        alias=assigned_user_alias  # Pass the aliased class instance
+    )
+]
+
+# Initialize your FastCRUD instance for TaskModel
+task_crud = FastCRUD(TaskModel)
+
+# Fetch tasks with joined user details
+tasks = await task_crud.get_multi_joined(
+    db=session,
+    schema_to_select=TaskSchema,
+    joins_config=joins_config,
+    offset=0,
+    limit=10
+)
+```
+
+Then just pass this joins_config to `get_multi_joined`:
+
+```python hl_lines="17" title="Passing joins_config to get_multi_joined"
+from fastcrud import FastCRUD, JoinConfig, aliased
+
+...
+
+# Configure joins with aliases
+joins_config = [
+    ...
+]
+
+# Initialize your FastCRUD instance for TaskModel
+task_crud = FastCRUD(TaskModel)
+
+# Fetch tasks with joined user details
+tasks = await task_crud.get_multi_joined(
+    db=session,
+    schema_to_select=TaskSchema,
+    joins_config=joins_config,
+    offset=0,
+    limit=10
+)
+```
+
+In this example, `owner_alias` and `assigned_user_alias` are created from `UserModel` to distinguish between the task's owner and the assigned user within the task management system. By using aliases, you can join the same model multiple times for different purposes in your queries, enhancing expressiveness and eliminating ambiguity.
+
 ## Conclusion
 
-The advanced features of FastCRUD, such as `allow_multiple` and support for advanced filters, empower developers to efficiently manage database records with complex conditions. By leveraging these capabilities, you can build more dynamic, robust, and scalable FastAPI applications that effectively interact with your data model.
+The advanced features of FastCRUD, such as `allow_multiple` and support for advanced filters, empower developers to efficiently manage database records with complex conditions. By leveraging these capabilities, you can build more dynamic, robust, and scalable FastAPI applications that effectively interact with your data model.
\ No newline at end of file
diff --git a/docs/advanced/overview.md b/docs/advanced/overview.md
index d820d64..8991f9e 100644
--- a/docs/advanced/overview.md
+++ b/docs/advanced/overview.md
@@ -24,5 +24,10 @@ This topic extends the use of `EndpointCreator` and `crud_router` for advanced e
 
 - [Advanced Endpoint Management Guide](endpoint.md#advanced-use-of-endpointcreator)
 
+### 5. Using `get_joined` and `get_multi_joined` for multiple models
+Explore the use of `get_joined` and `get_multi_joined` functions for complex queries that involve joining multiple models, including self-joins and scenarios requiring multiple joins on the same model.
+
+- [Joining Multiple Models Guide](crud.md#using-get_joined-and-get_multi_joined-for-multiple-models)
+
 ## Prerequisites
 Advanced usage assumes a solid understanding of the basic features and functionalities of our application. Knowledge of FastAPI, SQLAlchemy, and Pydantic is highly recommended to fully grasp the concepts discussed.
diff --git a/fastcrud/__init__.py b/fastcrud/__init__.py
index 6ba8e15..2f3a130 100644
--- a/fastcrud/__init__.py
+++ b/fastcrud/__init__.py
@@ -1,6 +1,16 @@
+from sqlalchemy.orm import aliased
+from sqlalchemy.orm.util import AliasedClass
+
 from .crud.fast_crud import FastCRUD
 from .endpoint.endpoint_creator import EndpointCreator
 from .endpoint.crud_router import crud_router
 from .crud.helper import JoinConfig
 
-__all__ = ["FastCRUD", "EndpointCreator", "crud_router", "JoinConfig"]
+__all__ = [
+    "FastCRUD",
+    "EndpointCreator",
+    "crud_router",
+    "JoinConfig",
+    "aliased",
+    "AliasedClass",
+]
diff --git a/fastcrud/crud/fast_crud.py b/fastcrud/crud/fast_crud.py
index a9aa444..c9c201a 100644
--- a/fastcrud/crud/fast_crud.py
+++ b/fastcrud/crud/fast_crud.py
@@ -9,13 +9,13 @@ from sqlalchemy.sql import Join
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.engine.row import Row
 from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm.util import AliasedClass
 from sqlalchemy.sql.elements import BinaryExpression
 from sqlalchemy.sql.selectable import Select
 
 from .helper import (
     _extract_matching_columns_from_schema,
     _auto_detect_join_condition,
-    _add_column_with_prefix,
     JoinConfig,
 )
 
@@ -537,12 +537,13 @@ class FastCRUD(
     async def get_joined(
         self,
         db: AsyncSession,
+        schema_to_select: Optional[type[BaseModel]] = None,
         join_model: Optional[type[DeclarativeBase]] = None,
-        join_prefix: Optional[str] = None,
         join_on: Optional[Union[Join, BinaryExpression]] = None,
-        schema_to_select: Optional[type[BaseModel]] = None,
+        join_prefix: Optional[str] = None,
         join_schema_to_select: Optional[type[BaseModel]] = None,
         join_type: str = "left",
+        alias: Optional[AliasedClass] = None,
         joins_config: Optional[list[JoinConfig]] = None,
         **kwargs: Any,
     ) -> Optional[dict[str, Any]]:
@@ -558,12 +559,13 @@ class FastCRUD(
 
         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.
             join_model: The model to join with.
-            join_prefix: Optional prefix to be added to all columns of the joined model. If None, no prefix is added.
             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.
-            schema_to_select: Pydantic schema for selecting specific columns from the primary model. Required if `return_as_model` is True.
+            join_prefix: Optional prefix to be added to all columns of the joined model. If None, no prefix is added.
             join_schema_to_select: Pydantic schema for selecting specific columns from the joined model.
             join_type: Specifies the type of join operation to perform. Can be "left" for a left outer join or "inner" for an inner join.
+            alias: An instance of `AliasedClass` for the join model, useful for self-joins or multiple joins on the same model. Result of `aliased(join_model)`.
             joins_config: A list of JoinConfig instances, each specifying a model to join with, join condition, optional prefix for column names, schema for selecting specific columns, and the type of join. This parameter enables support for multiple joins.
             **kwargs: Filters to apply to the primary model query, supporting advanced comparison operators for refined searching.
 
@@ -641,41 +643,50 @@ class FastCRUD(
             )
             ```
 
-            Return example: prefix added, no schema_to_select or join_schema_to_select
+            Using `alias` for joining the same model multiple times:
             ```python
-            {
-                "id": 1,
-                "name": "John Doe",
-                "username": "john_doe",
-                "email": "johndoe@example.com",
-                "hashed_password": "hashed_password_example",
-                "profile_image_url": "https://profileimageurl.com/default.jpg",
-                "uuid": "123e4567-e89b-12d3-a456-426614174000",
-                "created_at": "2023-01-01T12:00:00",
-                "updated_at": "2023-01-02T12:00:00",
-                "deleted_at": null,
-                "is_deleted": false,
-                "is_superuser": false,
-                "tier_id": 2,
-                "tier_name": "Premium",
-                "tier_created_at": "2022-12-01T10:00:00",
-                "tier_updated_at": "2023-01-01T11:00:00"
-            }
+            from fastcrud import aliased
+
+            owner_alias = aliased(ModelTest, name="owner")
+            user_alias = aliased(ModelTest, name="user")
+
+            result = await crud.get_joined(
+                db=session,
+                schema_to_select=BookingSchema,
+                joins_config=[
+                    JoinConfig(
+                        model=ModelTest,
+                        join_on=BookingModel.owner_id == owner_alias.id,
+                        join_prefix="owner_",
+                        alias=owner_alias,
+                        schema_to_select=UserSchema
+                    ),
+                    JoinConfig(
+                        model=ModelTest,
+                        join_on=BookingModel.user_id == user_alias.id,
+                        join_prefix="user_",
+                        alias=user_alias,
+                        schema_to_select=UserSchema
+                    )
+                ],
+                id=1
+            )
             ```
         """
         if joins_config and (
-            join_model or join_prefix or join_on or join_schema_to_select
+            join_model or join_prefix or join_on or join_schema_to_select or alias
         ):
             raise ValueError(
-                "Cannot use both single join parameters and joinsConfig simultaneously."
+                "Cannot use both single join parameters and joins_config simultaneously."
             )
         elif not joins_config and not join_model:
             raise ValueError("You need one of join_model or joins_config.")
 
         primary_select = _extract_matching_columns_from_schema(
-            model=self.model, schema=schema_to_select
+            model=self.model,
+            schema=schema_to_select,
         )
-        stmt: Select = select(*primary_select)
+        stmt: Select = select(*primary_select).select_from(self.model)
 
         join_definitions = joins_config if joins_config else []
         if join_model:
@@ -686,26 +697,24 @@ class FastCRUD(
                     join_prefix=join_prefix,
                     schema_to_select=join_schema_to_select,
                     join_type=join_type,
+                    alias=alias,
                 )
             )
 
         for join in join_definitions:
+            model = join.alias or join.model
+
             join_select = _extract_matching_columns_from_schema(
-                join.model, join.schema_to_select
+                model=join.model,
+                schema=join.schema_to_select,
+                prefix=join.join_prefix,
+                alias=join.alias,
             )
 
-            if join.join_prefix:
-                join_select = [
-                    _add_column_with_prefix(column, join.join_prefix)
-                    for column in join_select
-                ]
-
             if join.join_type == "left":
-                stmt = stmt.outerjoin(join.model, join.join_on).add_columns(
-                    *join_select
-                )
+                stmt = stmt.outerjoin(model, join.join_on).add_columns(*join_select)
             elif join.join_type == "inner":
-                stmt = stmt.join(join.model, join.join_on).add_columns(*join_select)
+                stmt = stmt.join(model, join.join_on).add_columns(*join_select)
             else:
                 raise ValueError(f"Unsupported join type: {join.join_type}.")
 
@@ -724,12 +733,13 @@ class FastCRUD(
     async def get_multi_joined(
         self,
         db: AsyncSession,
+        schema_to_select: Optional[type[BaseModel]] = None,
         join_model: Optional[type[ModelType]] = None,
-        join_prefix: Optional[str] = None,
         join_on: Optional[Any] = None,
-        schema_to_select: Optional[type[BaseModel]] = None,
+        join_prefix: Optional[str] = None,
         join_schema_to_select: Optional[type[BaseModel]] = None,
         join_type: str = "left",
+        alias: Optional[str] = None,
         offset: int = 0,
         limit: int = 100,
         sort_columns: Optional[Union[str, list[str]]] = None,
@@ -749,12 +759,13 @@ class FastCRUD(
 
         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.
             join_model: The model to join with.
-            join_prefix: Optional prefix to be added to all columns of the joined model. If None, no prefix is added.
             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.
-            schema_to_select: Pydantic schema for selecting specific columns from the primary model. Required if `return_as_model` is True.
+            join_prefix: Optional prefix to be added to all columns of the joined model. If None, no prefix is added.
             join_schema_to_select: Pydantic schema for selecting specific columns from the joined model.
             join_type: Specifies the type of join operation to perform. Can be "left" for a left outer join or "inner" for an inner join.
+            alias: An instance of `AliasedClass` for the join model, useful for self-joins or multiple joins on the same model. Result of `aliased(join_model)`.
             offset: The offset (number of records to skip) for pagination.
             limit: The limit (maximum number of records to return) for pagination.
             sort_columns: A single column name or a list of column names on which to apply sorting.
@@ -872,12 +883,49 @@ class FastCRUD(
                 sort_orders='asc'
             )
             ```
+
+            Example using `alias` for multiple joins, with pagination, sorting, and model conversion:
+            ```python
+            from fastcrud import JoinConfig, FastCRUD, aliased
+
+            # Aliasing for self-joins or multiple joins on the same table
+            owner_alias = aliased(ModelTest, name="owner")
+            user_alias = aliased(ModelTest, name="user")
+
+            # Initialize your FastCRUD instance for BookingModel
+            crud = FastCRUD(BookingModel)
+
+            result = await crud.get_multi_joined(
+                db=session,
+                schema_to_select=BookingSchema,  # Primary model schema
+                joins_config=[
+                    JoinConfig(
+                        model=ModelTest,
+                        join_on=BookingModel.owner_id == owner_alias.id,
+                        join_prefix="owner_",
+                        alias=owner_alias,
+                        schema_to_select=UserSchema  # Schema for the joined model
+                    ),
+                    JoinConfig(
+                        model=ModelTest,
+                        join_on=BookingModel.user_id == user_alias.id,
+                        join_prefix="user_",
+                        alias=user_alias,
+                        schema_to_select=UserSchema
+                    )
+                ],
+                offset=10,  # Skip the first 10 records
+                limit=5,  # Fetch up to 5 records
+                sort_columns=['booking_date'],  # Sort by booking_date
+                sort_orders=['desc']  # In descending order
+            )
+            ```
         """
         if joins_config and (
-            join_model or join_prefix or join_on or join_schema_to_select
+            join_model or join_prefix or join_on or join_schema_to_select or alias
         ):
             raise ValueError(
-                "Cannot use both single join parameters and joinsConfig simultaneously."
+                "Cannot use both single join parameters and joins_config simultaneously."
             )
         elif not joins_config and not join_model:
             raise ValueError("You need one of join_model or joins_config.")
@@ -885,9 +933,14 @@ class FastCRUD(
         if limit < 0 or offset < 0:
             raise ValueError("Limit and offset must be non-negative.")
 
-        joins: list[JoinConfig] = []
-        if join_model is not None:
-            joins.append(
+        primary_select = _extract_matching_columns_from_schema(
+            model=self.model, schema=schema_to_select
+        )
+        stmt: Select = select(*primary_select)
+
+        join_definitions = joins_config if joins_config else []
+        if join_model:
+            join_definitions.append(
                 JoinConfig(
                     model=join_model,
                     join_on=join_on
@@ -895,36 +948,24 @@ class FastCRUD(
                     join_prefix=join_prefix,
                     schema_to_select=join_schema_to_select,
                     join_type=join_type,
+                    alias=alias,
                 )
             )
-        elif joins_config:
-            joins.extend(joins_config)
-
-        primary_select = _extract_matching_columns_from_schema(
-            model=self.model, schema=schema_to_select
-        )
-        stmt: Select = select(*primary_select)
 
-        for join in joins:
-            if join.schema_to_select:
-                join_select = _extract_matching_columns_from_schema(
-                    join.model, join.schema_to_select
-                )
-            else:
-                join_select = inspect(join.model).c
+        for join in join_definitions:
+            model = join.alias or join.model
 
-            if join.join_prefix:
-                join_select = [
-                    _add_column_with_prefix(column, join.join_prefix)
-                    for column in join_select
-                ]
+            join_select = _extract_matching_columns_from_schema(
+                model=join.model,
+                schema=join.schema_to_select,
+                prefix=join.join_prefix,
+                alias=join.alias,
+            )
 
             if join.join_type == "left":
-                stmt = stmt.outerjoin(join.model, join.join_on).add_columns(
-                    *join_select
-                )
+                stmt = stmt.outerjoin(model, join.join_on).add_columns(*join_select)
             elif join.join_type == "inner":
-                stmt = stmt.join(join.model, join.join_on).add_columns(*join_select)
+                stmt = stmt.join(model, join.join_on).add_columns(*join_select)
             else:
                 raise ValueError(f"Unsupported join type: {join.join_type}.")
 
diff --git a/fastcrud/crud/helper.py b/fastcrud/crud/helper.py
index 308df74..63179bb 100644
--- a/fastcrud/crud/helper.py
+++ b/fastcrud/crud/helper.py
@@ -1,11 +1,9 @@
-from typing import Any, Union, Optional, NamedTuple
+from typing import Any, Optional, NamedTuple
+
 from sqlalchemy import inspect
-from sqlalchemy.orm import DeclarativeMeta
-from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import DeclarativeBase, DeclarativeMeta
+from sqlalchemy.orm.util import AliasedClass
 from sqlalchemy.sql import ColumnElement
-from sqlalchemy.sql.schema import Column
-from sqlalchemy.sql.elements import Label
-
 from pydantic import BaseModel
 
 
@@ -15,78 +13,48 @@ class JoinConfig(NamedTuple):
     join_prefix: Optional[str] = None
     schema_to_select: Optional[type[BaseModel]] = None
     join_type: str = "left"
+    alias: Optional[AliasedClass] = None
 
 
 def _extract_matching_columns_from_schema(
-    model: type[DeclarativeBase], schema: Optional[Union[type[BaseModel], list]]
+    model: type[DeclarativeBase],
+    schema: Optional[type[BaseModel]],
+    prefix: Optional[str] = None,
+    alias: Optional[AliasedClass] = None,
 ) -> list[Any]:
     """
-    Retrieves a list of ORM column objects from a SQLAlchemy model that match the field names in a given Pydantic schema.
+    Retrieves a list of ORM column objects from a SQLAlchemy model that match the field names in a given Pydantic schema,
+    or all columns from the model if no schema is provided. When an alias is provided, columns are referenced through
+    this alias, and a prefix can be applied to column names if specified.
 
     Args:
         model: The SQLAlchemy ORM model containing columns to be matched with the schema fields.
-        schema: The Pydantic schema containing field names to be matched with the model's columns.
-
-    Returns:
-        A list of ORM column objects from the model that correspond to the field names defined in the schema.
-    """
-    column_list = list(model.__table__.columns)
-    if schema is not None:
-        if isinstance(schema, list):
-            schema_fields = schema
-        else:
-            schema_fields = schema.model_fields.keys()
-
-        column_list = []
-        for column_name in schema_fields:
-            if hasattr(model, column_name):
-                column_list.append(getattr(model, column_name))
-
-    return column_list
-
-
-def _extract_matching_columns_from_kwargs(
-    model: type[DeclarativeBase], kwargs: dict[str, Any]
-) -> list[Any]:
-    """
-    Extracts matching ORM column objects from a SQLAlchemy model based on provided keyword arguments.
-
-    Args:
-        model: The SQLAlchemy ORM model.
-        kwargs: A dictionary containing field names as keys.
-
-    Returns:
-        A list of ORM column objects from the model that correspond to the field names provided in kwargs.
-    """
-    if kwargs is not None:
-        kwargs_fields = kwargs.keys()
-        column_list = []
-        for column_name in kwargs_fields:
-            if hasattr(model, column_name):
-                column_list.append(getattr(model, column_name))
-
-    return column_list
-
-
-def _extract_matching_columns_from_column_names(
-    model: type[DeclarativeBase], column_names: list[str]
-) -> list[Any]:
-    """
-    Extracts ORM column objects from a SQLAlchemy model based on a list of column names.
-
-    Args:
-        model: The SQLAlchemy ORM model.
-        column_names: A list of column names to extract.
+        schema: Optional; a Pydantic schema containing field names to be matched with the model's columns. If None, all columns from the model are used.
+        prefix: Optional; a prefix to be added to all column names. If None, no prefix is added.
+        alias: Optional; an alias for the model, used for referencing the columns through this alias in the query. If None, the original model is used.
 
     Returns:
-        A list of ORM column objects from the model that match the provided column names.
+        A list of ORM column objects (potentially labeled with a prefix) that correspond to the field names defined
+        in the schema or all columns from the model if no schema is specified. These columns are correctly referenced
+        through the provided alias if one is given.
     """
-    column_list = []
-    for column_name in column_names:
-        if hasattr(model, column_name):
-            column_list.append(getattr(model, column_name))
-
-    return column_list
+    model_or_alias = alias if alias else model
+    columns = []
+    if schema:
+        for field in schema.model_fields.keys():
+            if hasattr(model_or_alias, field):
+                column = getattr(model_or_alias, field)
+                if prefix:
+                    column = column.label(f"{prefix}{field}")
+                columns.append(column)
+    else:
+        for column in model.__table__.c:
+            column = getattr(model_or_alias, column.key)
+            if prefix:
+                column = column.label(f"{prefix}{column.key}")
+            columns.append(column)
+
+    return columns
 
 
 def _auto_detect_join_condition(
@@ -126,18 +94,3 @@ def _auto_detect_join_condition(
         )
 
     return join_on
-
-
-def _add_column_with_prefix(column: Column, prefix: Optional[str]) -> Label:
-    """
-    Creates a SQLAlchemy column label with an optional prefix.
-
-    Args:
-        column: The SQLAlchemy Column object to be labeled.
-        prefix: An optional prefix to prepend to the column's name.
-
-    Returns:
-        A labeled SQLAlchemy Column object.
-    """
-    column_label = f"{prefix}{column.name}" if prefix else column.name
-    return column.label(column_label)


Test patch:
diff --git a/tests/sqlalchemy/conftest.py b/tests/sqlalchemy/conftest.py
index c2a38c0..644fe89 100644
--- a/tests/sqlalchemy/conftest.py
+++ b/tests/sqlalchemy/conftest.py
@@ -1,4 +1,5 @@
 from typing import Optional
+from datetime import datetime
 
 import pytest
 import pytest_asyncio
@@ -45,6 +46,16 @@ class TierModel(Base):
     tests = relationship("ModelTest", back_populates="tier")
 
 
+class BookingModel(Base):
+    __tablename__ = "booking"
+    id = Column(Integer, primary_key=True)
+    owner_id = Column(Integer, ForeignKey("test.id"), nullable=False)
+    user_id = Column(Integer, ForeignKey("test.id"), nullable=False)
+    booking_date = Column(DateTime, nullable=False)
+    owner = relationship("ModelTest", foreign_keys=[owner_id], backref="owned_bookings")
+    user = relationship("ModelTest", foreign_keys=[user_id], backref="user_bookings")
+
+
 class CreateSchemaTest(BaseModel):
     model_config = ConfigDict(extra="forbid")
     name: str
@@ -80,6 +91,13 @@ class CategorySchemaTest(BaseModel):
     name: str
 
 
+class BookingSchema(BaseModel):
+    id: Optional[int] = None
+    owner_id: int
+    user_id: int
+    booking_date: datetime
+
+
 async_engine = create_async_engine(
     "sqlite+aiosqlite:///:memory:", echo=True, future=True
 )
@@ -137,6 +155,24 @@ def test_data_category() -> list[dict]:
     return [{"id": 1, "name": "Tech"}, {"id": 2, "name": "Health"}]
 
 
+@pytest.fixture(scope="function")
+def test_data_booking() -> list[dict]:
+    return [
+        {
+            "id": 1,
+            "owner_id": 1,
+            "user_id": 2,
+            "booking_date": datetime(2024, 3, 10, 15, 30),
+        },
+        {
+            "id": 2,
+            "owner_id": 1,
+            "user_id": 3,
+            "booking_date": datetime(2024, 3, 11, 10, 0),
+        },
+    ]
+
+
 @pytest.fixture
 def test_model():
     return ModelTest
diff --git a/tests/sqlalchemy/crud/test_get_joined.py b/tests/sqlalchemy/crud/test_get_joined.py
index 273c481..4ba581d 100644
--- a/tests/sqlalchemy/crud/test_get_joined.py
+++ b/tests/sqlalchemy/crud/test_get_joined.py
@@ -1,6 +1,6 @@
 import pytest
 from sqlalchemy import and_
-from fastcrud import FastCRUD, JoinConfig
+from fastcrud import FastCRUD, JoinConfig, aliased
 from ...sqlalchemy.conftest import (
     ModelTest,
     TierModel,
@@ -8,6 +8,9 @@ from ...sqlalchemy.conftest import (
     TierSchemaTest,
     CategoryModel,
     CategorySchemaTest,
+    BookingModel,
+    BookingSchema,
+    ReadSchemaTest,
 )
 
 
@@ -222,3 +225,113 @@ async def test_get_joined_multiple_models(
     assert "name" in result
     assert "tier_name" in result
     assert "category_name" in result
+
+
+@pytest.mark.asyncio
+async def test_get_joined_with_aliases(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    specific_booking_id = 1
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner = aliased(ModelTest, name="owner")
+    user = aliased(ModelTest, name="user")
+
+    result = await crud.get_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner.id,
+                join_prefix="owner_",
+                alias=owner,
+                schema_to_select=ReadSchemaTest,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user.id,
+                join_prefix="user_",
+                alias=user,
+                schema_to_select=ReadSchemaTest,
+            ),
+        ],
+        id=specific_booking_id,
+    )
+
+    assert result is not None
+    assert (
+        result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
+
+
+@pytest.mark.asyncio
+async def test_get_joined_with_aliases_no_schema(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    specific_booking_id = 1
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner = aliased(ModelTest, name="owner")
+    user = aliased(ModelTest, name="user")
+
+    result = await crud.get_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner.id,
+                join_prefix="owner_",
+                alias=owner,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user.id,
+                join_prefix="user_",
+                alias=user,
+            ),
+        ],
+        id=specific_booking_id,
+    )
+
+    assert result is not None
+    assert (
+        result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
diff --git a/tests/sqlalchemy/crud/test_get_multi_joined.py b/tests/sqlalchemy/crud/test_get_multi_joined.py
index 677f8d9..900ea2a 100644
--- a/tests/sqlalchemy/crud/test_get_multi_joined.py
+++ b/tests/sqlalchemy/crud/test_get_multi_joined.py
@@ -1,5 +1,5 @@
 import pytest
-from fastcrud import FastCRUD, JoinConfig
+from fastcrud import FastCRUD, JoinConfig, aliased
 from ...sqlalchemy.conftest import (
     ModelTest,
     TierModel,
@@ -8,6 +8,8 @@ from ...sqlalchemy.conftest import (
     ReadSchemaTest,
     CategoryModel,
     CategorySchemaTest,
+    BookingModel,
+    BookingSchema,
 )
 
 
@@ -304,3 +306,137 @@ async def test_get_multi_joined_with_additional_join_model(
     assert all(
         "tier_name" in item and "category_name" in item for item in result["data"]
     )
+
+
+@pytest.mark.asyncio
+async def test_get_multi_joined_with_aliases(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner_alias = aliased(ModelTest, name="owner")
+    user_alias = aliased(ModelTest, name="user")
+
+    result = await crud.get_multi_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner_alias.id,
+                join_prefix="owner_",
+                alias=owner_alias,
+                schema_to_select=ReadSchemaTest,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user_alias.id,
+                join_prefix="user_",
+                alias=user_alias,
+                schema_to_select=ReadSchemaTest,
+            ),
+        ],
+        offset=0,
+        limit=10,
+        sort_columns=["booking_date"],
+        sort_orders=["asc"],
+    )
+
+    assert "data" in result and isinstance(
+        result["data"], list
+    ), "The result should have a 'data' key with a list of records."
+    for booking in result["data"]:
+        assert (
+            "owner_name" in booking
+        ), "Each record should include 'owner_name' from the joined owner ModelTest data."
+        assert (
+            "user_name" in booking
+        ), "Each record should include 'user_name' from the joined user ModelTest data."
+    assert result is not None
+    assert result["total_count"] >= 1, "Expected at least one booking record"
+    first_result = result["data"][0]
+    assert (
+        first_result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        first_result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
+
+
+@pytest.mark.asyncio
+async def test_get_multi_joined_with_aliases_no_schema(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner_alias = aliased(ModelTest, name="owner")
+    user_alias = aliased(ModelTest, name="user")
+
+    result = await crud.get_multi_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner_alias.id,
+                join_prefix="owner_",
+                alias=owner_alias,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user_alias.id,
+                join_prefix="user_",
+                alias=user_alias,
+            ),
+        ],
+        offset=0,
+        limit=10,
+        sort_columns=["booking_date"],
+        sort_orders=["asc"],
+    )
+
+    assert "data" in result and isinstance(
+        result["data"], list
+    ), "The result should have a 'data' key with a list of records."
+    for booking in result["data"]:
+        assert (
+            "owner_name" in booking
+        ), "Each record should include 'owner_name' from the joined owner ModelTest data."
+        assert (
+            "user_name" in booking
+        ), "Each record should include 'user_name' from the joined user ModelTest data."
+    assert result is not None
+    assert result["total_count"] >= 1, "Expected at least one booking record"
+    first_result = result["data"][0]
+    assert (
+        first_result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        first_result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
diff --git a/tests/sqlmodel/conftest.py b/tests/sqlmodel/conftest.py
index 176d8c7..be5c066 100644
--- a/tests/sqlmodel/conftest.py
+++ b/tests/sqlmodel/conftest.py
@@ -47,6 +47,20 @@ class CreateSchemaTest(SQLModel):
     tier_id: int
 
 
+class BookingModel(SQLModel, table=True):
+    __tablename__ = "booking"
+    id: Optional[int] = Field(default=None, primary_key=True)
+    owner_id: int = Field(default=None, foreign_key="test.id")
+    user_id: int = Field(default=None, foreign_key="test.id")
+    booking_date: datetime
+    owner: "ModelTest" = Relationship(
+        sa_relationship_kwargs={"foreign_keys": "BookingModel.owner_id"}
+    )
+    user: "ModelTest" = Relationship(
+        sa_relationship_kwargs={"foreign_keys": "BookingModel.user_id"}
+    )
+
+
 class ReadSchemaTest(SQLModel):
     id: int
     name: str
@@ -69,6 +83,18 @@ class TierDeleteSchemaTest(SQLModel):
     pass
 
 
+class CategorySchemaTest(SQLModel):
+    id: Optional[int] = None
+    name: str
+
+
+class BookingSchema(SQLModel):
+    id: Optional[int] = None
+    owner_id: int
+    user_id: int
+    booking_date: datetime
+
+
 async_engine = create_async_engine(
     "sqlite+aiosqlite:///:memory:", echo=True, future=True
 )
@@ -102,17 +128,17 @@ async def async_session() -> AsyncSession:
 @pytest.fixture(scope="function")
 def test_data() -> list[dict]:
     return [
-        {"id": 1, "name": "Charlie", "tier_id": 1},
-        {"id": 2, "name": "Alice", "tier_id": 2},
-        {"id": 3, "name": "Bob", "tier_id": 1},
-        {"id": 4, "name": "David", "tier_id": 2},
-        {"id": 5, "name": "Eve", "tier_id": 1},
-        {"id": 6, "name": "Frank", "tier_id": 2},
-        {"id": 7, "name": "Grace", "tier_id": 1},
-        {"id": 8, "name": "Hannah", "tier_id": 2},
-        {"id": 9, "name": "Ivan", "tier_id": 1},
-        {"id": 10, "name": "Judy", "tier_id": 2},
-        {"id": 11, "name": "Alice", "tier_id": 1},
+        {"id": 1, "name": "Charlie", "tier_id": 1, "category_id": 1},
+        {"id": 2, "name": "Alice", "tier_id": 2, "category_id": 1},
+        {"id": 3, "name": "Bob", "tier_id": 1, "category_id": 2},
+        {"id": 4, "name": "David", "tier_id": 2, "category_id": 1},
+        {"id": 5, "name": "Eve", "tier_id": 1, "category_id": 1},
+        {"id": 6, "name": "Frank", "tier_id": 2, "category_id": 2},
+        {"id": 7, "name": "Grace", "tier_id": 1, "category_id": 2},
+        {"id": 8, "name": "Hannah", "tier_id": 2, "category_id": 1},
+        {"id": 9, "name": "Ivan", "tier_id": 1, "category_id": 1},
+        {"id": 10, "name": "Judy", "tier_id": 2, "category_id": 2},
+        {"id": 11, "name": "Alice", "tier_id": 1, "category_id": 1},
     ]
 
 
@@ -126,6 +152,24 @@ def test_data_category() -> list[dict]:
     return [{"id": 1, "name": "Tech"}, {"id": 2, "name": "Health"}]
 
 
+@pytest.fixture(scope="function")
+def test_data_booking() -> list[dict]:
+    return [
+        {
+            "id": 1,
+            "owner_id": 1,
+            "user_id": 2,
+            "booking_date": datetime(2024, 3, 10, 15, 30),
+        },
+        {
+            "id": 2,
+            "owner_id": 1,
+            "user_id": 3,
+            "booking_date": datetime(2024, 3, 11, 10, 0),
+        },
+    ]
+
+
 @pytest.fixture
 def test_model():
     return ModelTest
diff --git a/tests/sqlmodel/crud/test_get.py b/tests/sqlmodel/crud/test_get.py
index 75b01c6..63eae12 100644
--- a/tests/sqlmodel/crud/test_get.py
+++ b/tests/sqlmodel/crud/test_get.py
@@ -2,8 +2,8 @@ import pytest
 from pydantic import BaseModel
 
 from fastcrud.crud.fast_crud import FastCRUD
-from ...sqlalchemy.conftest import ModelTest
-from ...sqlalchemy.conftest import CreateSchemaTest
+from ...sqlmodel.conftest import ModelTest
+from ...sqlmodel.conftest import CreateSchemaTest
 
 
 @pytest.mark.asyncio
diff --git a/tests/sqlmodel/crud/test_get_joined.py b/tests/sqlmodel/crud/test_get_joined.py
index 273c481..32f765d 100644
--- a/tests/sqlmodel/crud/test_get_joined.py
+++ b/tests/sqlmodel/crud/test_get_joined.py
@@ -1,13 +1,16 @@
 import pytest
 from sqlalchemy import and_
-from fastcrud import FastCRUD, JoinConfig
-from ...sqlalchemy.conftest import (
+from fastcrud import FastCRUD, JoinConfig, aliased
+from ...sqlmodel.conftest import (
     ModelTest,
     TierModel,
     CreateSchemaTest,
     TierSchemaTest,
     CategoryModel,
     CategorySchemaTest,
+    BookingModel,
+    BookingSchema,
+    ReadSchemaTest,
 )
 
 
@@ -222,3 +225,113 @@ async def test_get_joined_multiple_models(
     assert "name" in result
     assert "tier_name" in result
     assert "category_name" in result
+
+
+@pytest.mark.asyncio
+async def test_get_joined_with_aliases(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    specific_booking_id = 1
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner = aliased(ModelTest, name="owner")
+    user = aliased(ModelTest, name="user")
+
+    result = await crud.get_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner.id,
+                join_prefix="owner_",
+                alias=owner,
+                schema_to_select=ReadSchemaTest,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user.id,
+                join_prefix="user_",
+                alias=user,
+                schema_to_select=ReadSchemaTest,
+            ),
+        ],
+        id=specific_booking_id,
+    )
+
+    assert result is not None
+    assert (
+        result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
+
+
+@pytest.mark.asyncio
+async def test_get_joined_with_aliases_no_schema(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    await async_session.commit()
+
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    specific_booking_id = 1
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner = aliased(ModelTest, name="owner")
+    user = aliased(ModelTest, name="user")
+
+    result = await crud.get_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner.id,
+                join_prefix="owner_",
+                alias=owner,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user.id,
+                join_prefix="user_",
+                alias=user,
+            ),
+        ],
+        id=specific_booking_id,
+    )
+
+    assert result is not None
+    assert (
+        result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
diff --git a/tests/sqlmodel/crud/test_get_multi_by_cursor.py b/tests/sqlmodel/crud/test_get_multi_by_cursor.py
index daa8e14..0ff57e6 100644
--- a/tests/sqlmodel/crud/test_get_multi_by_cursor.py
+++ b/tests/sqlmodel/crud/test_get_multi_by_cursor.py
@@ -1,6 +1,6 @@
 import pytest
 from fastcrud.crud.fast_crud import FastCRUD
-from ...sqlalchemy.conftest import ModelTest
+from ...sqlmodel.conftest import ModelTest
 
 
 @pytest.mark.asyncio
diff --git a/tests/sqlmodel/crud/test_get_multi_joined.py b/tests/sqlmodel/crud/test_get_multi_joined.py
index 0c1d812..4129f04 100644
--- a/tests/sqlmodel/crud/test_get_multi_joined.py
+++ b/tests/sqlmodel/crud/test_get_multi_joined.py
@@ -1,6 +1,6 @@
 import pytest
-from fastcrud import FastCRUD, JoinConfig
-from ...sqlalchemy.conftest import (
+from fastcrud import FastCRUD, JoinConfig, aliased
+from ...sqlmodel.conftest import (
     ModelTest,
     TierModel,
     CreateSchemaTest,
@@ -8,6 +8,8 @@ from ...sqlalchemy.conftest import (
     ReadSchemaTest,
     CategoryModel,
     CategorySchemaTest,
+    BookingModel,
+    BookingSchema,
 )
 
 
@@ -69,7 +71,6 @@ async def test_get_multi_joined_sorting(async_session, test_data, test_data_tier
 
 @pytest.mark.asyncio
 async def test_get_multi_joined_filtering(async_session, test_data, test_data_tier):
-    # Assuming there's a user with a specific name in test_data
     specific_user_name = "Charlie"
     for tier_item in test_data_tier:
         async_session.add(TierModel(**tier_item))
@@ -86,7 +87,7 @@ async def test_get_multi_joined_filtering(async_session, test_data, test_data_ti
         join_prefix="tier_",
         schema_to_select=CreateSchemaTest,
         join_schema_to_select=TierSchemaTest,
-        name=specific_user_name,  # Filter based on ModelTest attribute
+        name=specific_user_name,
         offset=0,
         limit=10,
     )
@@ -305,3 +306,137 @@ async def test_get_multi_joined_with_additional_join_model(
     assert all(
         "tier_name" in item and "category_name" in item for item in result["data"]
     )
+
+
+@pytest.mark.asyncio
+async def test_get_multi_joined_with_aliases(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner_alias = aliased(ModelTest, name="owner")
+    user_alias = aliased(ModelTest, name="user")
+
+    result = await crud.get_multi_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner_alias.id,
+                join_prefix="owner_",
+                alias=owner_alias,
+                schema_to_select=ReadSchemaTest,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user_alias.id,
+                join_prefix="user_",
+                alias=user_alias,
+                schema_to_select=ReadSchemaTest,
+            ),
+        ],
+        offset=0,
+        limit=10,
+        sort_columns=["booking_date"],
+        sort_orders=["asc"],
+    )
+
+    assert "data" in result and isinstance(
+        result["data"], list
+    ), "The result should have a 'data' key with a list of records."
+    for booking in result["data"]:
+        assert (
+            "owner_name" in booking
+        ), "Each record should include 'owner_name' from the joined owner ModelTest data."
+        assert (
+            "user_name" in booking
+        ), "Each record should include 'user_name' from the joined user ModelTest data."
+    assert result is not None
+    assert result["total_count"] >= 1, "Expected at least one booking record"
+    first_result = result["data"][0]
+    assert (
+        first_result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        first_result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
+
+
+@pytest.mark.asyncio
+async def test_get_multi_joined_with_aliases_no_schema(
+    async_session, test_data, test_data_tier, test_data_category, test_data_booking
+):
+    for tier_item in test_data_tier:
+        async_session.add(TierModel(**tier_item))
+    for category_item in test_data_category:
+        async_session.add(CategoryModel(**category_item))
+    for user_item in test_data:
+        async_session.add(ModelTest(**user_item))
+    for booking_item in test_data_booking:
+        async_session.add(BookingModel(**booking_item))
+    await async_session.commit()
+
+    crud = FastCRUD(BookingModel)
+
+    expected_owner_name = "Charlie"
+    expected_user_name = "Alice"
+
+    owner_alias = aliased(ModelTest, name="owner")
+    user_alias = aliased(ModelTest, name="user")
+
+    result = await crud.get_multi_joined(
+        db=async_session,
+        schema_to_select=BookingSchema,
+        joins_config=[
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.owner_id == owner_alias.id,
+                join_prefix="owner_",
+                alias=owner_alias,
+            ),
+            JoinConfig(
+                model=ModelTest,
+                join_on=BookingModel.user_id == user_alias.id,
+                join_prefix="user_",
+                alias=user_alias,
+            ),
+        ],
+        offset=0,
+        limit=10,
+        sort_columns=["booking_date"],
+        sort_orders=["asc"],
+    )
+
+    assert "data" in result and isinstance(
+        result["data"], list
+    ), "The result should have a 'data' key with a list of records."
+    for booking in result["data"]:
+        assert (
+            "owner_name" in booking
+        ), "Each record should include 'owner_name' from the joined owner ModelTest data."
+        assert (
+            "user_name" in booking
+        ), "Each record should include 'user_name' from the joined user ModelTest data."
+    assert result is not None
+    assert result["total_count"] >= 1, "Expected at least one booking record"
+    first_result = result["data"][0]
+    assert (
+        first_result["owner_name"] == expected_owner_name
+    ), "Owner name does not match expected value"
+    assert (
+        first_result["user_name"] == expected_user_name
+    ), "User name does not match expected value"
diff --git a/tests/sqlmodel/crud/test_update.py b/tests/sqlmodel/crud/test_update.py
index ad5bb41..15708cf 100644
--- a/tests/sqlmodel/crud/test_update.py
+++ b/tests/sqlmodel/crud/test_update.py
@@ -4,7 +4,7 @@ from sqlalchemy import select
 from sqlalchemy.exc import MultipleResultsFound
 
 from fastcrud.crud.fast_crud import FastCRUD
-from ...sqlalchemy.conftest import ModelTest
+from ...sqlmodel.conftest import ModelTest
 
 
 @pytest.mark.asyncio