feat(api): add permission-aware document count

This commit is contained in:
Julien Denizot
2026-06-20 15:48:51 +02:00
parent f63b78807d
commit 4d661e286a
2 changed files with 192 additions and 15 deletions
+91 -14
View File
@@ -287,20 +287,12 @@ def _parse_order_by(
return fieldname, direction
@frappe.whitelist(methods=["GET"])
def list_documents(
def _resolve_document_query_context(
doctype: str,
fields=None,
filters=None,
order_by_field: str | None = None,
order_direction: str = "desc",
limit_start: int | str = 0,
limit_page_length: int | str = DEFAULT_PAGE_LENGTH,
) -> dict[str, Any]:
) -> tuple[str, str, Any, str, set[str]]:
"""
Return a permission-aware, read-only page of exposed documents.
Only explicitly requested and validated fields are returned.
Resolve and validate the common security context used by document
read-only endpoints.
"""
user = frappe.session.user
@@ -329,13 +321,13 @@ def list_documents(
if meta.istable:
frappe.throw(
_("Child-table documents cannot be listed directly."),
_("Child-table documents cannot be queried directly."),
frappe.ValidationError,
)
if meta.issingle:
frappe.throw(
_("Single DocTypes cannot be listed."),
_("Single DocTypes cannot be queried."),
frappe.ValidationError,
)
@@ -370,6 +362,33 @@ def list_documents(
user=user,
)
return user, doctype_name, meta, app, allowed_fields
@frappe.whitelist(methods=["GET"])
def list_documents(
doctype: str,
fields=None,
filters=None,
order_by_field: str | None = None,
order_direction: str = "desc",
limit_start: int | str = 0,
limit_page_length: int | str = DEFAULT_PAGE_LENGTH,
) -> dict[str, Any]:
"""
Return a permission-aware, read-only page of exposed documents.
Only explicitly requested and validated fields are returned.
"""
(
user,
doctype_name,
meta,
app,
allowed_fields,
) = _resolve_document_query_context(doctype)
selected_fields = _parse_fields(
fields,
allowed_fields=allowed_fields,
@@ -440,3 +459,61 @@ def list_documents(
for row in page
],
}
@frappe.whitelist(methods=["GET"])
def count_documents(
doctype: str,
filters=None,
) -> dict[str, Any]:
"""
Return the exact permission-aware count of exposed documents.
The aggregation is fixed by the Bridge. Callers cannot provide
arbitrary SQL expressions or aggregation fields.
"""
(
user,
doctype_name,
meta,
app,
allowed_fields,
) = _resolve_document_query_context(doctype)
validated_filters = _parse_filters(
filters,
allowed_fields=allowed_fields,
)
rows = frappe.get_list(
doctype_name,
fields=[
{
"COUNT": "name",
"as": "count",
}
],
filters=validated_filters,
limit_page_length=1,
user=user,
)
count = 0
if rows:
count = max(cint(rows[0].get("count")), 0)
return {
"api_version": "v1",
"user": user,
"read_only": True,
"doctype": doctype_name,
"app": app,
"module": meta.module,
"permission_basis": "frappe_get_list",
"exposure_basis": "enuxia_site_policy",
"filters_count": len(validated_filters),
"count": count,
}
+101 -1
View File
@@ -3,7 +3,10 @@ from unittest.mock import patch
import frappe
from frappe.tests.utils import FrappeTestCase
from enuxia_ai_bridge.api.documents import list_documents
from enuxia_ai_bridge.api.documents import (
count_documents,
list_documents,
)
from enuxia_ai_bridge.exposure import (
ExposureDecision,
ExposureEffect,
@@ -222,3 +225,100 @@ class TestDocumentList(FrappeTestCase):
self.assertIsNone(
result["pagination"]["next_limit_start"]
)
class TestDocumentCount(FrappeTestCase):
def setUp(self) -> None:
frappe.set_user("Administrator")
def tearDown(self) -> None:
frappe.set_user("Administrator")
def test_requires_authentication(self) -> None:
frappe.set_user("Guest")
with self.assertRaises(frappe.AuthenticationError):
count_documents("Customer")
def test_returns_permission_aware_count(self) -> None:
with (
patch(
"enuxia_ai_bridge.api.documents."
"_resolve_document_query_context",
return_value=(
"Administrator",
"Customer",
frappe._dict(module="Selling"),
"erpnext",
{"name", "disabled"},
),
),
patch(
"enuxia_ai_bridge.api.documents."
"frappe.get_list",
return_value=[
frappe._dict(count=2),
],
) as get_list,
):
result = count_documents(
"Customer",
filters={"disabled": 0},
)
self.assertEqual(result["count"], 2)
self.assertEqual(result["doctype"], "Customer")
self.assertEqual(result["filters_count"], 1)
self.assertTrue(result["read_only"])
self.assertEqual(
result["permission_basis"],
"frappe_get_list",
)
kwargs = get_list.call_args.kwargs
self.assertEqual(
kwargs["fields"],
[
{
"COUNT": "name",
"as": "count",
}
],
)
self.assertEqual(
kwargs["filters"],
{"disabled": 0},
)
self.assertEqual(
kwargs["limit_page_length"],
1,
)
self.assertEqual(
kwargs["user"],
"Administrator",
)
def test_returns_zero_when_aggregation_is_empty(self) -> None:
with (
patch(
"enuxia_ai_bridge.api.documents."
"_resolve_document_query_context",
return_value=(
"Administrator",
"Customer",
frappe._dict(module="Selling"),
"erpnext",
{"name", "disabled"},
),
),
patch(
"enuxia_ai_bridge.api.documents."
"frappe.get_list",
return_value=[],
),
):
result = count_documents("Customer")
self.assertEqual(result["count"], 0)