gitea-admin/nocobase-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: add fixed project write service

Browse Source
This commit is contained in:
nac
2026-07-13 01:47:43 +00:00
parent 4ab02025c7
commit 6eff1aba59
2 changed files with 383 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nocobase_crm_mcp/projects.py
+223
View File
@@ -0,0 +1,223 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from datetime import date
from typing import Any
from .config import ProjectConfig
from .nocobase_http import NocoBaseHttpError
def _configured_scalar_fields(config: ProjectConfig) -> tuple[str, ...]:
return (
config.name_field,
config.customer_field,
config.inquiry_date_field,
config.progress_field,
config.applications_field,
config.part_numbers_field,
config.period_usage_field,
config.details_field,
)
PROJECT_WHITELIST = _configured_scalar_fields(ProjectConfig())
class _InvalidProjectInput(ValueError):
pass
class _InvalidProgress(ValueError):
pass
class ProjectService:
def __init__(self, config: ProjectConfig, nb_client: Any, http_client: Any) -> None:
self.config = config
self.nb_client = nb_client
self.http_client = http_client
self._whitelist = _configured_scalar_fields(config)
def create_project(
self,
*,
name: str,
customer_id: Any | None = None,
inquiry_date: str | None = None,
progress: str | None = None,
applications: str | None = None,
part_numbers: str | None = None,
period_usage: str | None = None,
details: str | None = None,
attachment_ids: Sequence[Any] | None = None,
) -> dict[str, Any]:
try:
values = self._project_values(
name=name,
customer_id=customer_id,
inquiry_date=inquiry_date,
progress=progress,
applications=applications,
part_numbers=part_numbers,
period_usage=period_usage,
details=details,
require_name=True,
)
attachment_ids = _deduplicated_attachment_ids(attachment_ids)
except _InvalidProgress:
return _error("invalid_progress", "progress must be one of the configured project progress labels")
except _InvalidProjectInput as error:
return _error("invalid_project_input", str(error))
response = self.nb_client.create_record(self.config.resource, values, self._whitelist)
project_id = _project_id(response, self.config.id_field)
if not attachment_ids:
return {"project_id": project_id}
try:
self.http_client.add_attachment_ids(project_id, attachment_ids)
except Exception as error:
return _attachment_error(project_id, error)
return {"project_id": project_id}
def update_project(
self,
project_id: Any,
*,
name: str | None = None,
customer_id: Any | None = None,
inquiry_date: str | None = None,
progress: str | None = None,
applications: str | None = None,
part_numbers: str | None = None,
period_usage: str | None = None,
details: str | None = None,
add_attachment_ids: Sequence[Any] | None = None,
remove_attachment_ids: Sequence[Any] | None = None,
) -> dict[str, Any]:
try:
values = self._project_values(
name=name,
customer_id=customer_id,
inquiry_date=inquiry_date,
progress=progress,
applications=applications,
part_numbers=part_numbers,
period_usage=period_usage,
details=details,
require_name=False,
)
add_attachment_ids = _deduplicated_attachment_ids(add_attachment_ids)
remove_attachment_ids = _deduplicated_attachment_ids(remove_attachment_ids)
if _overlap(add_attachment_ids, remove_attachment_ids):
raise _InvalidProjectInput("attachment IDs cannot be added and removed in the same request")
if not values and not add_attachment_ids and not remove_attachment_ids:
raise _InvalidProjectInput("update request must include project fields or attachment IDs")
except _InvalidProgress:
return _error("invalid_progress", "progress must be one of the configured project progress labels")
except _InvalidProjectInput as error:
return _error("invalid_project_input", str(error))
if values:
self.nb_client.update_record(self.config.resource, project_id, values, self._whitelist)
try:
if add_attachment_ids:
self.http_client.add_attachment_ids(project_id, add_attachment_ids)
if remove_attachment_ids:
self.http_client.remove_attachment_ids(project_id, remove_attachment_ids)
except Exception as error:
return _attachment_error(project_id, error)
return {"project_id": project_id}
def _project_values(
self,
*,
name: str | None,
customer_id: Any | None,
inquiry_date: str | None,
progress: str | None,
applications: str | None,
part_numbers: str | None,
period_usage: str | None,
details: str | None,
require_name: bool,
) -> dict[str, Any]:
values: dict[str, Any] = {}
if name is not None:
if not isinstance(name, str) or not (normalized_name := name.strip()):
raise _InvalidProjectInput("name must not be blank")
values[self.config.name_field] = normalized_name
elif require_name:
raise _InvalidProjectInput("name is required")
if customer_id is not None:
values[self.config.customer_field] = customer_id
if inquiry_date is not None:
if not isinstance(inquiry_date, str):
raise _InvalidProjectInput("inquiry_date must be a YYYY-MM-DD date")
try:
parsed_date = date.fromisoformat(inquiry_date)
if parsed_date.isoformat() != inquiry_date:
raise ValueError("date must use YYYY-MM-DD format")
values[self.config.inquiry_date_field] = parsed_date.isoformat()
except ValueError as error:
raise _InvalidProjectInput("inquiry_date must be a YYYY-MM-DD date") from error
if progress is not None:
if not isinstance(progress, str) or progress not in self.config.progress_values:
raise _InvalidProgress
values[self.config.progress_field] = self.config.progress_values[progress]
for value, field in (
(applications, self.config.applications_field),
(part_numbers, self.config.part_numbers_field),
(period_usage, self.config.period_usage_field),
(details, self.config.details_field),
):
if value is not None:
values[field] = value
return values
def _deduplicated_attachment_ids(attachment_ids: Sequence[Any] | None) -> list[Any]:
if attachment_ids is None:
return []
if isinstance(attachment_ids, (str, bytes)) or not isinstance(attachment_ids, Sequence):
raise _InvalidProjectInput("attachment IDs must be a list")
unique_ids: list[Any] = []
for attachment_id in attachment_ids:
if attachment_id is None or isinstance(attachment_id, (Mapping, Sequence)):
raise _InvalidProjectInput("attachment IDs must be scalar values")
if attachment_id not in unique_ids:
unique_ids.append(attachment_id)
return unique_ids
def _overlap(left: Sequence[Any], right: Sequence[Any]) -> bool:
return any(left_id == right_id for left_id in left for right_id in right)
def _project_id(response: Mapping[str, Any], id_field: str) -> Any:
data = response.get("data")
if not isinstance(data, Mapping) or id_field not in data:
raise ValueError(f"NocoBase create response did not contain {id_field!r}")
return data[id_field]
def _attachment_error(project_id: Any, error: Exception) -> dict[str, Any]:
if isinstance(error, NocoBaseHttpError):
if error.status_code == 404:
code = "attachment_not_found"
elif error.status_code in {401, 403}:
code = "nb_upload_unauthorized"
else:
code = "nb_upload_failed"
else:
code = "nb_upload_failed"
return {"project_id": project_id, "error": {"code": code, "message": str(error)}}
def _error(code: str, message: str) -> dict[str, Any]:
return {"error": {"code": code, "message": message}}
tests/test_projects.py
+160
View File
@@ -0,0 +1,160 @@
import unittest
from nocobase_crm_mcp.config import ProjectConfig
from nocobase_crm_mcp.nocobase_http import NocoBaseHttpError
from nocobase_crm_mcp.projects import PROJECT_WHITELIST, ProjectService
class FakeNbClient:
def __init__(self) -> None:
self.create_calls: list[tuple[str, dict, tuple[str, ...]]] = []
self.update_calls: list[tuple[str, object, dict, tuple[str, ...]]] = []
def create_record(self, resource, values, whitelist):
self.create_calls.append((resource, dict(values), tuple(whitelist)))
return {"data": {"id": 41}}
def update_record(self, resource, project_id, values, whitelist):
self.update_calls.append((resource, project_id, dict(values), tuple(whitelist)))
return {"data": {"id": project_id}}
class FakeHttpClient:
def __init__(self) -> None:
self.add_calls: list[tuple[object, list[object]]] = []
self.remove_calls: list[tuple[object, list[object]]] = []
self.add_error: Exception | None = None
self.remove_error: Exception | None = None
def add_attachment_ids(self, project_id, attachment_ids) -> None:
self.add_calls.append((project_id, list(attachment_ids)))
if self.add_error:
raise self.add_error
def remove_attachment_ids(self, project_id, attachment_ids) -> None:
self.remove_calls.append((project_id, list(attachment_ids)))
if self.remove_error:
raise self.remove_error
class ProjectServiceTests(unittest.TestCase):
def setUp(self) -> None:
self.nb = FakeNbClient()
self.http = FakeHttpClient()
self.config = ProjectConfig()
self.service = ProjectService(self.config, self.nb, self.http)
def test_create_maps_fixed_fields_then_adds_attachment_ids(self) -> None:
result = self.service.create_project(
name=" EV inverter ", customer_id=3, progress="SOP", attachment_ids=[8, 9]
)
self.assertEqual(
self.nb.create_calls,
[
(
"t_projects",
{
"f_project_name": "EV inverter",
"client_ID": 3,
"f_projects_progress": "a9t9awhtt3r",
},
PROJECT_WHITELIST,
)
],
)
self.assertEqual(self.http.add_calls, [(41, [8, 9])])
self.assertEqual(result["project_id"], 41)
def test_whitelist_contains_only_the_configured_scalar_fields(self) -> None:
self.service.create_project(name="x")
self.assertEqual(
PROJECT_WHITELIST,
(
self.config.name_field,
self.config.customer_field,
self.config.inquiry_date_field,
self.config.progress_field,
self.config.applications_field,
self.config.part_numbers_field,
self.config.period_usage_field,
self.config.details_field,
),
)
self.assertNotIn(self.config.attachment_field, PROJECT_WHITELIST)
def test_update_rejects_overlapping_attachment_lists_before_writing(self) -> None:
result = self.service.update_project(41, add_attachment_ids=[8], remove_attachment_ids=[8])
self.assertEqual(result["error"]["code"], "invalid_project_input")
self.assertEqual(self.nb.update_calls, [])
self.assertEqual(self.http.add_calls, [])
self.assertEqual(self.http.remove_calls, [])
def test_invalid_date_progress_and_blank_name_have_stable_codes(self) -> None:
self.assertEqual(
self.service.create_project(name="x", inquiry_date="2026/07/12")["error"]["code"],
"invalid_project_input",
)
self.assertEqual(
self.service.create_project(name="x", inquiry_date="20260712")["error"]["code"],
"invalid_project_input",
)
self.assertEqual(
self.service.create_project(name="x", progress="草稿")["error"]["code"],
"invalid_progress",
)
self.assertEqual(self.service.create_project(name=" \t ")["error"]["code"], "invalid_project_input")
def test_update_rejects_an_empty_request_before_writing(self) -> None:
result = self.service.update_project(41)
self.assertEqual(result["error"]["code"], "invalid_project_input")
self.assertEqual(self.nb.update_calls, [])
def test_update_writes_scalars_then_deduplicated_attachment_changes_in_order(self) -> None:
result = self.service.update_project(
41,
inquiry_date="2026-07-12",
add_attachment_ids=[8, 8, 9],
remove_attachment_ids=[10, 10],
)
self.assertEqual(
self.nb.update_calls,
[
(
"t_projects",
41,
{"xm_project_inquiry_date": "2026-07-12"},
PROJECT_WHITELIST,
)
],
)
self.assertEqual(self.http.add_calls, [(41, [8, 9])])
self.assertEqual(self.http.remove_calls, [(41, [10])])
self.assertEqual(result, {"project_id": 41})
def test_association_errors_keep_project_id_and_map_to_stable_codes(self) -> None:
cases = [
(404, "attachment_not_found"),
(401, "nb_upload_unauthorized"),
(403, "nb_upload_unauthorized"),
(500, "nb_upload_failed"),
]
for status_code, error_code in cases:
with self.subTest(status_code=status_code):
self.setUp()
self.http.add_error = NocoBaseHttpError(status_code, "attachment request failed")
result = self.service.create_project(name="x", attachment_ids=[8])
self.assertEqual(result["project_id"], 41)
self.assertEqual(result["error"]["code"], error_code)
self.assertEqual(len(self.nb.create_calls), 1)
if __name__ == "__main__":
unittest.main()