Suggestion : Search with dynamic parameters

Makefile

@@ -16,7 +16,8 @@ run_dev:
 	git ls-files | entr -r	pipenv run python dev.py
 
 tdd:
-	git ls-files | entr  make test functionnal_tests opt='$(opt)'
+	git ls-files | entr  make test opt='$(opt)'
+	git ls-files | entr make functionnal_tests
 
 refactor_tdd:
 	make tdd opt="--pdb --ff --lf --ff -x"

crud.py

@@ -37,14 +37,28 @@ def get_movie_by_name(db: Session, name: str = ""):
     return db_movie.all()
 
 
-def get_all_movies(db: Session, offset: int | None = None, limit: int | None = None):
+def get_all_movies(db: Session):
     db_movie = db.query(models.Movie)
-    if offset is not None:
-        db_movie = db_movie.offset(offset)
-    if limit is not None:
-        db_movie = db_movie.limit(limit)
+    return db_movie.all()
 
-    return db_movie
+
+def search_movie(db: Session, term: str = "", *criterions):
+    db_movies = db.query(models.Movie)
+    exp = f"%{term}%"
+    # Use regex instead ? still \W is had to escape
+    if term:
+        db_movies = db_movies.where(
+            sqlalchemy.or_(
+                models.Movie.title.like(exp),
+                models.Movie.description.like(exp),
+            )
+        )
+    # would try the following approache for dynamic search
+
+    for colnamename, opname, value in criterions:
+        db_movies = db_movies.where(getattr(getattr(models.Movie, name), opname)(value))
+
+    return db_movies
 
 
 def get_movie_by_id(db: Session, id_: str = ""):

dev.py

@@ -114,40 +114,11 @@ async def delete_movie(id_: str, db: Session = Depends(get_db)) -> None:
 
 
 @app.get("/movies/")
-async def list_movie(
-    db: Session = Depends(get_db),
-    pagenum: int | None = None,
-    pagesize: int | None = None,
-) -> schemas.PaginatedMovies | schemas.MovieObjectsOut:
-    paginate_params = {}
-    paginate_data = {}
+async def list_movie(db: Session = Depends(get_db)) -> schemas.MovieObjectsOut:
+    movies = crud.get_all_movies(db)
+    count = len(movies)
 
-    pagination_params = {"pagenum": pagenum, "pagesize": pagesize}
-    if any(v for v in pagination_params.values() if v is not None):
-        missing = [name for (name, value) in pagination_params.items() if not value]
-        if missing:
-            raise HTTPException(status_code=404, detail=f"Missing {missing}")
-
-        # Here we do a "x + 1 - 1 = x" trick to check if there will be more pages
-        # eg we want from 10 to 15, we ask for 10 to 16, if we have *stricly* more
-        # than 5 element we can now that there will be one more page
-        paginate_params = dict(offset=(pagenum - 1) * pagesize, limit=pagesize + 1)
-
-    movies = crud.get_all_movies(db, **paginate_params)
-
-    if paginate_params:
-        has_more_content = movies.count() > pagesize
-        paginate_data = {
-            "next_page": pagenum + 1 if has_more_content else None,
-            "previous_page": pagenum - 1 if pagenum > 1 else None,
-        }
-
-        movies = movies.limit(pagesize)
-
-    count = movies.count()
-
-    payload = {"movies": movies, "count": count}
-    return {**payload, **paginate_data}
+    return {"movies": movies, "count": count}
 
 
 if __name__ == "__main__":

schemas.py

@@ -16,15 +16,6 @@ class MovieObject(MoviePayload):
     id: int | str
 
 
-class Paginated(BaseModel):
-    next_page: str | int | None
-    previous_page: str | int | None
-
-
 class MovieObjectsOut(BaseModel):
     movies: list[MovieObject]
     count: int
-
-
-class PaginatedMovies(MovieObjectsOut, Paginated):
-    pass

utests/test_api.py

@@ -157,130 +157,6 @@ class BaseCrud(unittest.TestCase):
         assert isinstance(movies["movies"], list)
         assert movies["count"] == primary_count + N
 
-    def test_list_pagination_limits(self):
-        response = client.get("/movies/")
-        nb_movies = response.json()["count"]
-
-        for _ in range(3):
-            self.create_payload["title"] = rand_name()
-            response = client.post("/movies/", json=self.create_payload)
-
-        response = client.get("/movies/")
-        nb_movies = response.json()["count"]
-
-        pagenum = 1
-        pagesize = nb_movies - 1
-
-        # Test page 1 has no previous ?
-        current_movies = client.get(
-            f"/movies/?pagenum={pagenum}&pagesize={pagesize}"
-        ).json()
-        assert current_movies.get("previous_page") is None
-        assert current_movies["next_page"]
-
-        current_movies = client.get(
-            f"/movies/?pagenum={pagenum + 1 }&pagesize={pagesize}"
-        ).json()
-        assert current_movies.get("next_page") is None
-        assert current_movies["previous_page"]
-
-        # test last page has no next
-
-    def test_list_movies_pagination_back_forth(self):
-        response = client.get("/movies/")
-        nb_movies = response.json()["count"]
-
-        for _ in range(3):
-            self.create_payload["title"] = rand_name()
-            response = client.post("/movies/", json=self.create_payload)
-
-        response = client.get("/movies/")
-        nb_movies = response.json()["count"]
-
-        pagenum = 1
-        pagesize = 2
-
-        first, *_, last = client.get("/movies/").json()["movies"]
-
-        while current_movies := client.get(
-            f"/movies/?pagenum={pagenum}&pagesize={pagesize}"
-        ).json():
-            next_page_num = current_movies.get("next_page")
-            assert next_page_num != pagenum
-
-            if next_page_num is None:
-                assert current_movies["movies"][-1] == last
-                break
-            else:
-                assert next_page_num == pagenum + 1
-                pagenum = next_page_num
-
-    def test_list_movies_pagination(self):
-        response = client.get("/movies/")
-        assert response.status_code == 200
-        # assert response.json() == []
-        primary_count = response.json()["count"]
-
-        N = 10
-        names = []
-        for _ in range(N):
-            name = rand_name()
-
-            names.append(name)
-            self.create_payload["title"] = name
-            response = client.post("/movies/", json=self.create_payload)
-            assert response.status_code == 200
-
-        pagenum = 3
-        pagesize = 5
-
-        sliced_movies = client.get("/movies/").json()["movies"][
-            (pagenum - 1) * pagesize : pagenum * pagesize
-        ]
-
-        sliced_titles = [m["title"] for m in sliced_movies]
-
-        movies_paginate = client.get(
-            f"/movies/?pagenum={pagenum}&pagesize={pagesize}"
-        ).json()["movies"]
-
-        paginate_titles = [m["title"] for m in movies_paginate]
-
-        assert sliced_titles == paginate_titles
-
-    def test_list_movies_pagination(self):
-        response = client.get("/movies/")
-        assert response.status_code == 200
-        # assert response.json() == []
-        primary_count = response.json()["count"]
-
-        N = 10
-        names = []
-        for _ in range(N):
-            name = rand_name()
-
-            names.append(name)
-            self.create_payload["title"] = name
-            response = client.post("/movies/", json=self.create_payload)
-            assert response.status_code == 200
-
-        pagenum = 3
-        pagesize = 5
-
-        sliced_movies = client.get("/movies/").json()["movies"][
-            (pagenum - 1) * pagesize : pagenum * pagesize
-        ]
-
-        sliced_titles = [m["title"] for m in sliced_movies]
-
-        movies_paginate = client.get(
-            f"/movies/?pagenum={pagenum}&pagesize={pagesize}"
-        ).json()["movies"]
-
-        paginate_titles = [m["title"] for m in movies_paginate]
-
-        assert sliced_titles == paginate_titles
-
 
 class ApiTestCase(unittest.TestCase):
     def test_payload_content_in_and_out_loopback(self):

utests/test_sql_database.py

@@ -122,34 +122,88 @@ def test_list_movies():
     assert all(movies_by_title[name] for name in names)
 
 
-def test_search_movies():
+def test_search_movies_exact():
     clear_db()
     response = client.get("/movies/")
     # assert response.json() == []
 
     radix = rand_name()
 
-    name = radix + "test_search"
+    title = radix + "test_search_title"
 
-    desc = radix + "test_desription"
+    desc = radix + "test_search_desription"
 
     with db_context() as db:
-        movie_name = crud.create_movie(
-            db, title=name, genres=["Animated", "Paropaganda"]
+        movie_title = crud.create_movie(
+            db, title=title, genres=["Animated", "Paropaganda"]
         )
+
         movie_desc = crud.create_movie(
             db, title=radix, description=desc, genres=["Animated", "Paropaganda"]
         )
+        movie_desc_id = movie_desc.id
 
-    for term, target in zip((name, desc), (movie_name, movie_desc)):
-        with db_context() as db:
-            found = crud.search_movie(db, desc).all()
-            assert len(found) == 1
-            assert target == found[0]
+        found = crud.search_movie(db, title).all()
+        assert len(found) == 1
 
-    movies = client.get("movies").json()["movies"]
-    movies_by_title = {m["title"]: m for m in movies}
-    assert all(movies_by_title[name] for name in names)
+        assert found[0] == movie_title
+
+        found = crud.search_movie(db, desc).all()
+        assert len(found) == 1
+
+        assert found[0] == movie_desc
+
+
+def test_search_movies_token():
+    clear_db()
+    response = client.get("/movies/")
+    # assert response.json() == []
+
+    radix = rand_name()
+
+    title = radix + "test_search_title titletoken"
+
+    desc = radix + "test_search_desription desctoken"
+
+    with db_context() as db:
+        movie_title = crud.create_movie(
+            db, title=title, genres=["Animated", "Paropaganda"]
+        )
+
+        movie_desc = crud.create_movie(
+            db, title=radix, description=desc, genres=["Animated", "Paropaganda"]
+        )
+        movie_desc_id = movie_desc.id
+
+        found = crud.search_movie(db, "titletoken").all()
+        assert len(found) == 1
+
+        assert found[0] == movie_title
+
+        found = crud.search_movie(db, "desctoken").all()
+        assert len(found) == 1
+
+        assert found[0] == movie_desc
+
+
+def test_search_movies_token_period():
+    return
+    clear_db()
+    response = client.get("/movies/")
+    # assert response.json() == []
+
+    radix = rand_name()
+
+    title = radix + "test_search_title title3tokenperiod."
+
+    with db_context() as db:
+        movie_title = crud.create_movie(
+            db, title=title, genres=["Animated", "Paropaganda"]
+        )
+
+        found = crud.search_movie(db, "title3tokenpreriod").all()
+        assert len(found) == 1
+        assert found[0] == movie_title
 
 
 def test_sample_import_toy_story():