Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions application/database/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -884,6 +884,34 @@ def get_by_tags(self, tags: List[str]) -> List[cre_defs.Document]:
)
return documents

def get_by_tags_with_pagination(
self, tags: List[str], page: int = 1, items_per_page: int = 20
) -> Tuple[Optional[int], List[cre_defs.Document]]:

if not tags:
return 0, []

# Get all documents matching tags (reuse existing logic)
all_documents = self.get_by_tags(tags)

# Apply manual pagination since we have mixed types
total_items = len(all_documents)
total_pages = (
total_items + items_per_page - 1
) // items_per_page # Ceiling division

if total_pages == 0:
return 0, []

# Calculate slice indices
start_idx = (page - 1) * items_per_page
end_idx = start_idx + items_per_page

# Return paginated slice
paginated_documents = all_documents[start_idx:end_idx]

return total_pages, paginated_documents

def get_nodes_with_pagination(
self,
name: str,
Expand Down
48 changes: 48 additions & 0 deletions application/tests/db_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,54 @@ def test_get_by_tags(self) -> None:
self.assertEqual(self.collection.get_by_tags([]), [])
self.assertEqual(self.collection.get_by_tags(["this should not be a tag"]), [])

def test_get_by_tags_with_pagination(self) -> None:

for i in range(5):
dbcre = db.CRE(
description=f"Pagiation CRE {i}",
name=f"PaginationCRE{i}",
tags="pagination-test",
external_id=f"500-{i:03d}",
)
self.collection.session.add(dbcre)
self.collection.session.commit()

total_pages, docs = self.collection.get_by_tags_with_pagination(
["pagination-test"], page=1, items_per_page=10
)
self.assertEqual(len(docs), 5)
self.assertEqual(total_pages, 1)

total_pages, docs = self.collection.get_by_tags_with_pagination(
["pagination-test"], page=1, items_per_page=2
)
self.assertEqual(len(docs), 2)
self.assertEqual(total_pages, 3)

total_pages, docs = self.collection.get_by_tags_with_pagination(
["pagination-test"], page=2, items_per_page=2
)
self.assertEqual(len(docs), 2)
self.assertEqual(total_pages, 3)

total_pages, docs = self.collection.get_by_tags_with_pagination(
["pagination-test"], page=3, items_per_page=2
)
self.assertEqual(len(docs), 1)
self.assertEqual(total_pages, 3)

total_pages, docs = self.collection.get_by_tags_with_pagination(
[], page=1, items_per_page=1
)
self.assertEqual(total_pages, 0)
self.assertEqual(docs, [])

total_pages, docs = self.collection.get_by_tags_with_pagination(
["non-existent-tag"], page=1, items_per_page=10
)
self.assertEqual(total_pages, 0)
self.assertEqual(docs, [])

def test_get_standards_names(self) -> None:
result = self.collection.get_node_names()
expected = [("Standard", "BarStand"), ("Standard", "Unlinked")]
Expand Down
6 changes: 5 additions & 1 deletion application/tests/web_main_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,7 +400,11 @@ def test_find_document_by_tag(self) -> None:
response = client.get(f"/rest/v1/tags?tag=CW")
self.assertEqual(404, response.status_code)

expected = {"data": [cres["ca"].todict(), cres["cb"].todict()]}
expected = {
"data": [cres["ca"].todict(), cres["cb"].todict()],
"total_pages": 1,
"page": 1,
}

response = client.get(f"/rest/v1/tags?tag=ta")
self.assertEqual(200, response.status_code)
Expand Down
20 changes: 15 additions & 5 deletions application/web/web_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -242,16 +242,26 @@ def find_node_by_name(
@app.route("/rest/v1/tags", methods=["GET"])
def find_document_by_tag() -> Any:
tags = request.args.getlist("tag")
opt_format = request.args.get("format")

page = 1
if request.args.get("page") is not None and int(request.args.get("page")) > 0:
page = int(request.args.get("page"))

items_per_page = int(request.args.get("items_per_page") or ITEMS_PER_PAGE)

if posthog:
posthog.capture(f"find_document_by_tag", f"tags:{tags}")
posthog.capture(f"find_document_by_tag", f"tags:{tags};page:{page}")

database = db.Node_collection()
# opt_osib = request.args.get("osib")
opt_format = request.args.get("format")
documents = database.get_by_tags(tags)
total_pages, documents = database.get_by_tags_with_pagination(
tags, page=page, items_per_page=items_per_page
)

if documents:
res = [doc.todict() for doc in documents]
result = {"data": res}
result = {"data": res, "total_pages": total_pages, "page": page}

# if opt_osib:
# result["osib"] = odefs.cre2osib(documents).todict()
if opt_format == SupportedFormats.Markdown.value:
Expand Down