Remove curried methods and toolz library

pyfastapi/repositories/base.py

@@ -1,6 +1,5 @@
 from abc import ABC
-from typing import Annotated, Tuple, List, Type
-from toolz.functoolz import curry  # type: ignore
+from typing import Annotated, Tuple, List, Type, TypeVar
 
 from fastapi import Depends
 from pydantic import BaseModel
@@ -11,19 +10,17 @@
 from pyfastapi.models.base import Base
 from pyfastapi.schemas.base import BaseEnum
 
+T = TypeVar("T", bound=Base)
+
 
 class BaseRepository(ABC):
     def __init__(self, db: Annotated[Session, Depends(get_db)]):
         self.db = db
 
 
-@curry  # type: ignore
 def extract_query(
-    model: Type[BaseModel], use_like_list: List[str], q: BaseModel, stmt: Select[Tuple[Base]]
-) -> Select[Tuple[Base]]:
-    """
-    because of the @curry, return type is being ignored and is using Any
-    """
+    model: Type[T], use_like_list: List[str], q: BaseModel, stmt: Select[Tuple[T]]
+) -> Select[Tuple[T]]:
     stmt_ = stmt
     for attr, value in q:
         if value is not None:
@@ -34,23 +31,18 @@ def extract_query(
     return stmt_
 
 
-@curry  # type: ignore
 def extract_sort(
-    model: Type[BaseModel], enum: Type[BaseEnum], sort: str, stmt: Select[Tuple[Type[Base]]]
-) -> Select[Tuple[Type[Base]]]:
-    """
-    because of the @curry, return type is being ignored and is using Any
-    """
-    if sort:
-        stmt_ = stmt
-        first_char = sort[0]
-        sort_key = sort[1:] if first_char in ["-", "+"] else sort
-
-        if sort_key in enum:
-            if sort[0] == "-":
-                stmt_ = stmt_.order_by(getattr(model, sort_key).desc())
-            else:
-                stmt_ = stmt_.order_by(getattr(model, sort_key).asc())
-        return stmt_
-    else:
+    model: Type[T], enum: Type[BaseEnum], sort: str, stmt: Select[Tuple[T]]
+) -> Select[Tuple[T]]:
+    if not sort:
+        return stmt
+
+    first_char = sort[0]
+    sort_key = sort[1:] if first_char in ["-", "+"] else sort
+
+    if sort_key not in enum:
         return stmt
+
+    if sort[0] == "-":
+        return stmt.order_by(getattr(model, sort_key).desc())
+    return stmt.order_by(getattr(model, sort_key).asc())

pyfastapi/repositories/country.py

@@ -1,4 +1,3 @@
-from toolz.functoolz import compose  # type: ignore
 from fastapi_pagination import LimitOffsetPage
 from fastapi_pagination.ext.sqlalchemy import paginate
 from sqlalchemy.orm import joinedload
@@ -15,10 +14,8 @@ def get_country(self, code: str) -> Country | None:
         return self.db.execute(stmt).scalar_one_or_none()
 
     def get_countries(self, q: QueryCountrySchema, sort: str) -> LimitOffsetPage[Country]:
-        f = compose(
-            extract_query(Country, ["name"], q),
-            extract_sort(Country, SortCountryEnum, sort)
-        )
-        stmt = f(select(Country))
+        stmt = select(Country)
+        stmt = extract_sort(Country, SortCountryEnum, sort, stmt)
+        stmt = extract_query(Country, ["name"], q, stmt)
         country_list: LimitOffsetPage[Country] = paginate(self.db, stmt)
         return country_list

pyfastapi/repositories/person.py

@@ -1,4 +1,3 @@
-from toolz .functoolz import compose  # type: ignore
 from fastapi_pagination import LimitOffsetPage
 from fastapi_pagination.ext.sqlalchemy import paginate
 from sqlalchemy.dialects.sqlite import insert
@@ -16,11 +15,9 @@ def get_person(self, id_: int) -> Person | None:
         return self.db.execute(stmt).scalar_one_or_none()
 
     def get_persons(self, q: QueryPersonSchema, sort: str) -> LimitOffsetPage[Person]:
-        f = compose(
-            extract_query(Person, ["first_name", "last_name"], q),
-            extract_sort(Person, SortPersonEnum, sort)
-        )
-        stmt = f(select(Person))
+        stmt = select(Person)
+        stmt = extract_sort(Person, SortPersonEnum, sort, stmt)
+        stmt = extract_query(Person, ["first_name", "last_name"], q, stmt)
         person_list: LimitOffsetPage[Person] = paginate(self.db, stmt)
         return person_list
 

pyproject.toml

@@ -10,7 +10,6 @@ dependencies = [
     "fastapi-pagination>=0.12.14,<0.13",
     "sqlalchemy>=2.0.25,<3",
     "alembic>=1.13.1,<2",
-    "toolz>=0.12.1,<0.13",
 ]
 
 [dependency-groups]

tests/unit_tests/repositories/test_base.py

@@ -23,24 +23,21 @@ def setUpClass(cls) -> None:
         cls.repo = CountryRepository(Session())
 
     def test_country(self) -> None:
-        val = extract_sort(Country, SortCountryEnum, "name")(select(Country))
+        val = extract_sort(Country, SortCountryEnum, "name", select(Country))
         print("val", val)
 
     def test_person(self) -> None:
-        val = extract_sort(Person, SortPersonEnum, "id")(select(Person))
+        val = extract_sort(Person, SortPersonEnum, "id", select(Person))
         print("person", val)
 
 
 class TestCountryRepository(TestCase):
     def generate_from_sqlalchemy(self, sort_key_: str) -> str:
-        stmt_ = extract_sort(Country, SortCountryEnum, sort_key_)(select(Country))
+        stmt_ = extract_sort(Country, SortCountryEnum, sort_key_, select(Country))
         return str(stmt_)
 
     def get_extract_query(self, q: QueryCountrySchema) -> Select[Tuple[Country]]:
-        """
-        ignore because it returns any
-        """
-        return extract_query(Country, ["name"], q)(select(Country))  # type: ignore
+        return extract_query(Country, ["name"], q, select(Country))
 
     def test_country_extract_sort_if_part_of_enum_asc(self) -> None:
         def generate_sql(sort_key_: str) -> str:
@@ -95,14 +92,11 @@ def test_country_extract_query_not_in_enum(self) -> None:
 
 class TestPersonRepository(TestCase):
     def generate_from_sqlalchemy(self, sort_key_: str) -> str:
-        stmt_ = extract_sort(Person, SortPersonEnum, sort_key_)(select(Person))
+        stmt_ = extract_sort(Person, SortPersonEnum, sort_key_, select(Person))
         return str(stmt_)
 
     def get_extract_query(self, q: QueryPersonSchema) -> Select[Tuple[Person]]:
-        """
-        ignore because it returns any
-        """
-        return extract_query(Person, ["first_name", "last_name"], q)(select(Person))  # type: ignore
+        return extract_query(Person, ["first_name", "last_name"], q, select(Person))
 
     # extract sort
     def test_person_extract_sort_if_part_of_enum_asc(self) -> None: