From 7cfc1693d639c19e43b4c3d41c74992ed434ab79 Mon Sep 17 00:00:00 2001 From: jingrow Date: Sat, 25 Oct 2025 03:46:42 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0api=5Fadapter.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/jingrow/jingrow/adapters/api_adapter.py | 192 +++++++++++++++---- 1 file changed, 154 insertions(+), 38 deletions(-) diff --git a/apps/jingrow/jingrow/adapters/api_adapter.py b/apps/jingrow/jingrow/adapters/api_adapter.py index ae2abc5..c5aa03b 100644 --- a/apps/jingrow/jingrow/adapters/api_adapter.py +++ b/apps/jingrow/jingrow/adapters/api_adapter.py @@ -180,78 +180,194 @@ class ApiAdapter: return [self._create_page(pagetype, item.pop('name', None), **item) for item in data] - def new_pg(self, pagetype: str, **kwargs): - """创建新的页面类型文档""" - logger.info(f"API new_pg: {pagetype}") - return self._create_page(pagetype, None, **kwargs) - - def save_pg(self, pg): - """保存页面类型文档""" - logger.info(f"API save_pg: {pg.pagetype}, {pg.name}") + def new_pg(self, pagetype: str, *, parent_pg=None, parentfield=None, as_dict=False, **kwargs): + """创建新的页面类型文档 - 对齐云端 new_pg 接口 - endpoint = self._build_endpoint(pg.pagetype, pg.name) - method = 'PUT' if pg.name else 'POST' + :param pagetype: PageType of the new document. + :param parent_pg: [optional] add to parent document. + :param parentfield: [optional] add against this `parentfield`. + :param as_dict: [optional] return as dictionary instead of Page. + :param kwargs: [optional] You can specify fields as field=value pairs in function call. + """ + logger.info(f"API new_pg: {pagetype}") + + # 构建新文档数据 + new_data = { + "pagetype": pagetype, + "__islocal": 1, + "pagestatus": 0, + **kwargs + } + + # 如果有父文档，添加父文档信息 + if parent_pg: + new_data["parent"] = parent_pg.name if hasattr(parent_pg, 'name') else str(parent_pg) + new_data["parenttype"] = parent_pg.pagetype if hasattr(parent_pg, 'pagetype') else None + if parentfield: + new_data["parentfield"] = parentfield + + if as_dict: + return new_data + else: + # 移除 pagetype 避免参数冲突 + pagetype_data = new_data.pop('pagetype') + return self._create_page(pagetype_data, None, **new_data) + + def save_pg(self, pg, ignore_permissions=False, ignore_validate=False, ignore_mandatory=False): + """保存页面类型文档 - 对齐云端 save_pg 接口 + + :param pg: Page object to save + :param ignore_permissions: Ignore permission checks + :param ignore_validate: Ignore validation + :param ignore_mandatory: Ignore mandatory field checks + """ + logger.info(f"API save_pg: {pg.pagetype}, {getattr(pg, 'name', None)}") + + endpoint = self._build_endpoint(pg.pagetype, getattr(pg, 'name', None)) + method = 'PUT' if getattr(pg, 'name', None) else 'POST' data = pg.as_dict() + # 添加保存选项 + if ignore_permissions: + data['ignore_permissions'] = True + if ignore_validate: + data['ignore_validate'] = True + if ignore_mandatory: + data['ignore_mandatory'] = True + result = self._make_request(method, endpoint, data) # 更新文档对象 if result: - pg._data.update(result) + # 直接更新对象的属性 + for key, value in result.items(): + setattr(pg, key, value) return pg - def delete_pg(self, pagetype: str, name: str): - """删除页面类型文档""" + def delete_pg(self, pagetype: str, name: str, ignore_missing=False): + """删除页面类型文档 - 对齐云端 delete_pg 接口 + + :param pagetype: PageType name + :param name: Document name + :param ignore_missing: Ignore if document doesn't exist + """ logger.info(f"API delete_pg: {pagetype}, {name}") endpoint = self._build_endpoint(pagetype, name) - self._make_request('DELETE', endpoint) + params = {'ignore_missing': ignore_missing} if ignore_missing else {} + + self._make_request('DELETE', endpoint, params) return True - def db_exists(self, pagetype: str, filters: Dict[str, Any] = None): - """检查文档是否存在""" + def db_exists(self, pagetype: str, filters: Dict[str, Any] = None, cache=False): + """检查文档是否存在 - 对齐云端 db.exists 接口 + + :param pagetype: PageType name + :param filters: Filter conditions (dict or name string) + :param cache: Enable caching for single document checks + + Examples: + - db_exists("User", "jane@example.org", cache=True) + - db_exists("User", {"full_name": "Jane Doe"}) + - db_exists({"pagetype": "User", "full_name": "Jane Doe"}) + """ logger.info(f"API db_exists: {pagetype}, {filters}") + # 处理字典格式的 pagetype + if isinstance(pagetype, dict): + filters = pagetype.copy() + pagetype = filters.pop("pagetype") + + # 如果 filters 是字符串，转换为字典 + if isinstance(filters, str): + filters = {"name": filters} + endpoint = self._build_endpoint(pagetype) params = {'filters': filters} if filters else {} + if cache: + params['cache'] = cache data = self._make_request('GET', endpoint, params) return len(data) > 0 - def db_get_value(self, pagetype: str, filters: Dict[str, Any] = None, fieldname: str = None): - """获取数据库值""" + def db_get_value(self, pagetype: str, filters: Dict[str, Any] = None, fieldname: str = None, + ignore=False, cache=False, order_by=None, parent=None): + """获取数据库值 - 对齐云端 db.get_value 接口 + + :param pagetype: PageType name + :param filters: Filter conditions (dict or name string) + :param fieldname: Field name to get value for + :param ignore: Ignore if document doesn't exist + :param cache: Enable caching + :param order_by: Order by field + :param parent: Parent document for child table queries + + Examples: + - db_get_value("User", "jane@example.org", "full_name") + - db_get_value("User", {"email": "jane@example.org"}, "full_name") + """ logger.info(f"API db_get_value: {pagetype}, {filters}, {fieldname}") + # 如果 filters 是字符串，转换为字典 + if isinstance(filters, str): + filters = {"name": filters} + endpoint = self._build_endpoint(pagetype) - params = {'filters': filters, 'fieldname': fieldname} if filters else {} + params = { + 'filters': filters, + 'fieldname': fieldname + } + if ignore: + params['ignore'] = ignore + if cache: + params['cache'] = cache + if order_by: + params['order_by'] = order_by + if parent: + params['parent'] = parent data = self._make_request('GET', endpoint, params) return data - def db_set_value(self, pagetype: str, filters: Dict[str, Any] = None, fieldname: str = None, value: Any = None): - """设置数据库值""" + def db_set_value(self, pagetype: str, filters: Dict[str, Any] = None, fieldname: str = None, + value: Any = None, modified=None, modified_by=None, update_modified=True, debug=False): + """设置数据库值 - 对齐云端 db.set_value 接口 + + :param pagetype: PageType name + :param filters: Filter conditions (dict or name string) + :param fieldname: Field name to set value for + :param value: Value to set + :param modified: Use this as the modified timestamp + :param modified_by: Set this user as modified_by + :param update_modified: Update the modified timestamp + :param debug: Debug mode + + Examples: + - db_set_value("User", "jane@example.org", "full_name", "Jane Doe") + - db_set_value("User", {"email": "jane@example.org"}, "full_name", "Jane Doe") + """ logger.info(f"API db_set_value: {pagetype}, {filters}, {fieldname}, {value}") + # 如果 filters 是字符串，转换为字典 + if isinstance(filters, str): + filters = {"name": filters} + endpoint = self._build_endpoint(pagetype) - data = {'filters': filters, 'fieldname': fieldname, 'value': value} + data = { + 'filters': filters, + 'fieldname': fieldname, + 'value': value + } + if modified: + data['modified'] = modified + if modified_by: + data['modified_by'] = modified_by + if not update_modified: + data['update_modified'] = update_modified + if debug: + data['debug'] = debug self._make_request('PUT', endpoint, data) return True - def log_error(self, title: str, message: str): - """记录错误日志""" - logger.error(f"[{title}] {message}") - - # 也可以发送到 SaaS 版 - try: - endpoint = "api/method/jingrow.core.utils.log_error" - data = {'title': title, 'message': message} - self._make_request('POST', endpoint, data) - except Exception as e: - logger.warning(f"Failed to log error to SaaS: {e}") - - def throw(self, title: str, message: str): - """抛出异常""" - from ..exceptions import JingrowException - raise JingrowException(title, message)