Implement Case Sensitive Queries using SQL Alchemy

[juaansb01]

🚀 Feature Request

Is your feature request related to a problem? Please describe.

• Using this archetype, in one of the projects in which we use it, the need has arisen to make a strict check between upper and lower case words. I have not been able to get the current methods to allow this strict validation. That is why I propose to introduce the functionality.

---

Describe the solution you'd like

• As I say, I would like the get_one_by_field``, get_one_by_fieldsandget_listmethods, that appear insrc/repository/crud/base.py```, which are the ones that use values that can be received as input, in my case it is through the JSON body of the API, to be able to execute DB queries making them case sensitive.

---

Describe alternatives you've considered

• As this casuistry is the first time it appears, I think it would be good if the function could let you decide whether or not you want the methods to fulfil this function.

---

Additional context

• I leave here the code for the methods that we have written to achieve this functionality. I'm open to any comments or improvements you see to make it as good as possible.

def get_one_by_field(
        self: "CRUDBase[ModelType]",
        db: Session,
        field: str,
        value: str,
        case_sensitive: bool = False,  # noqa: FBT001, FBT002
    ) -> ModelType:
        """Returns an object of the model specified, with optional case sensitivity.

        Args:
            db (Session): Database session.
            field (str): Field of the row in the DB.
            value (Any): Value to compare the Field with.
            case_sensitive (bool): If True, applies case-sensitive filtering. Defaults to False.

        Returns:
            ModelType: Element.

        Raises:
            ElementNotFoundError: If the element is not found.
        """
        logger.info("Entering...")
        logger.debug("Getting %s with %s: %s", self.model.__name__, field, value)

        filter_field = getattr(self.model, field)
        if isinstance(value, str) and case_sensitive:
            filter_field = filter_field.collate("utf8mb4_bin")

        query = db.query(self.model).filter(filter_field == value)

        if data := query.first():
            logger.debug("Found %s with %s: %s", self.model.__name__, field, value)
            logger.info("Exiting...")
            return data

        error_msg = f"{self.model.__name__} with {field}: {value} not found."
        logger.error(error_msg)
        logger.info("Exiting...")
        raise ElementNotFoundError(error_msg)

    def get_one_by_fields(
        self: "CRUDBase[ModelType]",
        db: Session,
        filters: list[Filter],
    ) -> ModelType:
        """Returns an object of the model specified.

        Args:
            db (Session): Database session.
            filters (dict[str, Tuple[str, object]]): Filters to apply, where each filter
                is a tuple of (operator, value).

        Returns:
            ModelType: Element.

        Raises:
            ElementNotFoundError: If the element is not found.
        """
        logger.info("Entering...")
        logger.debug("Getting %s with filters: %s", self.model.__name__, filters)
        filter_clauses = self._get_filters(filters)
        if data := db.query(self.model).filter(*filter_clauses).first():
            logger.debug("Found %s with filters: %s", self.model.__name__, filters)
            logger.info("Exiting...")
            return data
        error_msg = f"{self.model.__name__} with filters: {filters} not found."
        logger.error(error_msg)
        logger.info("Exiting...")
        raise ElementNotFoundError(error_msg)

    def get_list(  # noqa: C901
        self: "CRUDBase[ModelType]",
        db: Session,
        offset: int | None = None,
        limit: int | None = None,
        filters: list[Filter] | None = None,
        filter_is_logic_and: bool = True,  # noqa: FBT002, FBT001
        order_by: str = "id",
        order_direction: Literal["asc", "desc"] = "asc",  # noqa: ARG002
        join_fields: list[str] | None = None,
        case_sensitive: bool = False,  # noqa: FBT002, FBT001
    ) -> Sequence[ModelType | None]:
        """Get a list of elements that can be filtered, with optional case sensitivity.

        Result requires mapping the objects to the desired response.

        Args:
            db (Session): Database session.
            offset (int | None = None): Omit a specified number of rows before
                the beginning of the result set. Defaults to None.
            limit (int | None = None): Limit the number of rows returned from a query.
                Defaults to None.
            filters (dict[str, Tuple[str, object]], optional): Filters to apply, where each filter
                is a tuple of (operator, value). Defaults to None.
            filter_is_logic_and (bool, optional): If True, the filters are applied with AND logic,
                otherwise with OR logic. Defaults to True.
            order_by (str, optional): Field to order the results by. Defaults to "id".
            order_direction (Literal["asc", "desc"], optional): Order direction for the results.
            join_fields (list[str], optional): List of foreign key fields to perform
                joined loading on. Defaults to None.
            case_sensitive (bool): If True, applies case-sensitive filtering. Defaults to False.

        Returns:
            list[ModelType | None]: Result with the Data.
        """
        logger.info("Entering...")
        logger.debug("Getting list of %s", self.model.__name__)

        query = select(self.model)

        if join_fields:
            for join_field in join_fields:
                query = query.join(getattr(self.model, join_field))

        if filters:
            filter_clauses = []
            for filter_obj in filters:
                field_parts = filter_obj.field.split(".")
                filter_field = getattr(self.model, field_parts[0])

                for part in field_parts[1:]:
                    filter_field = getattr(filter_field.property.mapper.class_, part)

                if isinstance(filter_obj.value, str) and case_sensitive:
                    # Use COLLATE for case-sensitive comparison
                    filter_field = filter_field.collate("utf8mb4_bin")

                filter_clauses.append(
                    self._get_filter_expression(filter_field, filter_obj.operator, filter_obj.value)
                )

            if filter_is_logic_and:
                query = query.where(*filter_clauses)
            else:
                query = query.filter(or_(*filter_clauses))
            logger.debug("Filters applied: %s", filters)

        # Apply ordering
        # order_field = getattr(self.model, order_by)
        # query = query.order_by(
        #     order_field.desc() if order_direction == "desc" else order_field.asc()
        # )
        logger.debug("Order by: %s", order_by)

        # Apply offset and limit - Pagination
        if offset:
            query = query.offset(offset)
            logger.debug("Offset: %s", offset)
        if limit:
            query = query.limit(limit)
            logger.debug("Limit: %s", limit)

        string_query = str(query)
        logger.debug("Query: %s", string_query)

        if data := db.scalars(query).all():
            logger.debug("Found list of %s", self.model.__name__)
            logger.info("Exiting...")
            return data

        logger.error("List of %s not found", self.model.__name__)
        logger.info("Exiting...")
        return []

• We were using MySQL 8.0. The string type columns had charset utf8mb4 and collation utf8mb4_0900_ai_ci.