Compare commits

..

No commits in common. "pagination" and "main" have entirely different histories.

5 changed files with 8 additions and 174 deletions

View File

@ -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"

View File

@ -37,14 +37,9 @@ 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
return db_movie.all()
def get_movie_by_id(db: Session, id_: str = ""):

37
dev.py
View File

@ -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__":

View File

@ -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

View File

@ -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):