From 040269478a0c48adce18bbb7c2d0e97ab9efded4 Mon Sep 17 00:00:00 2001 From: jingrow Date: Sat, 25 Oct 2025 03:22:44 +0800 Subject: [PATCH] =?UTF-8?q?get=5Fpg=E5=92=8Cget=5Flist=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E4=BA=91=E7=AB=AF=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/jingrow/jingrow/__init__.py | 4 + apps/jingrow/jingrow/adapters/api_adapter.py | 153 +++++++++++++------ 2 files changed, 114 insertions(+), 43 deletions(-) diff --git a/apps/jingrow/jingrow/__init__.py b/apps/jingrow/jingrow/__init__.py index 0d107bd..228712e 100644 --- a/apps/jingrow/jingrow/__init__.py +++ b/apps/jingrow/jingrow/__init__.py @@ -53,6 +53,10 @@ def log_error(title: str, message: str): def throw(title: str, message: str): return _adapter.throw(title, message) +def _dict(): + """创建一个空字典""" + return {} + # =============== Whitelist 实现 =============== _logger = logging.getLogger(__name__) _whitelisted_functions: Dict[str, Dict[str, Any]] = {} diff --git a/apps/jingrow/jingrow/adapters/api_adapter.py b/apps/jingrow/jingrow/adapters/api_adapter.py index 30ce1dc..ae2abc5 100644 --- a/apps/jingrow/jingrow/adapters/api_adapter.py +++ b/apps/jingrow/jingrow/adapters/api_adapter.py @@ -63,65 +63,133 @@ class ApiAdapter: logger.error(f"API request failed: {e}") raise - def get_pg(self, pagetype: str, name: str = None, **kwargs): - """获取页面类型文档""" - logger.info(f"API get_pg: {pagetype}, {name}") - + def _create_page(self, pagetype: str, name: str = None, **data): + """创建 Page 对象的工厂方法""" + page_data = {"pagetype": pagetype, **data} + if name is not None: + page_data["name"] = name + return Page(page_data) + + def _build_endpoint(self, pagetype: str, name: str = None) -> str: + """构建 API 端点""" endpoint = f"api/data/{pagetype}" if name: endpoint += f"/{name}" + return endpoint + + def get_pg(self, *args, **kwargs): + """获取页面类型文档 - 对齐云端 get_pg 接口 + 支持多种调用方式： + - get_pg("PageType", "name") - 获取指定文档 + - get_pg({"pagetype": "PageType", ...}) - 从字典创建文档 + - get_pg(pagetype="PageType", field=...) - 关键字参数方式 + - get_pg("PageType", "name", for_update=True) - 带更新锁 + """ + pagetype = None + name = None + + # 处理参数 + if args: + if isinstance(args[0], str): + pagetype = args[0] + if len(args) > 1: + name = args[1] + elif isinstance(args[0], dict): + # 从字典创建文档 + kwargs = args[0] + else: + raise ValueError("First non keyword argument must be a string or dict") + + # 从 kwargs 中提取 pagetype + if pagetype is None and kwargs: + if "pagetype" in kwargs: + pagetype = kwargs["pagetype"] + else: + raise ValueError('"pagetype" is a required key') + + logger.info(f"API get_pg: {pagetype}, {name}") + + # 如果只有 pagetype 没有 name，返回列表 + if name is None: + # 从 kwargs 中移除 pagetype，避免参数冲突 + list_kwargs = kwargs.copy() + list_kwargs.pop('pagetype', None) + return self.get_list(pagetype, **list_kwargs) + + # 获取单个文档 + endpoint = self._build_endpoint(pagetype, name) data = self._make_request('GET', endpoint, kwargs) - # 转换为 Page 对象 - class MockPage(Page): - def __init__(self, pagetype, name=None, **kwargs): - d = {"pagetype": pagetype, **kwargs} - if name is not None: - d["name"] = name - super().__init__(d) + # 如果 API 返回的数据中有 name，先移除它，使用我们指定的 name + if isinstance(data, dict) and 'name' in data: + api_name = data.pop('name') - return MockPage(pagetype, name, **data) + return self._create_page(pagetype, name, **data) - def get_list(self, pagetype: str, *args, **kwargs): - """获取页面类型列表""" + def get_list(self, pagetype: str, fields=None, filters=None, group_by=None, + order_by=None, limit_start=None, limit_page_length=20, + parent=None, debug=False, as_dict=True, or_filters=None, **kwargs): + """获取页面类型列表 - 对齐云端 get_list 接口 + + :param pagetype: PageType on which query is to be made. + :param fields: List of fields or `*`. + :param filters: List of filters (see example). + :param order_by: Order By e.g. `modified desc`. + :param limit_start: Start results at record #. Default 0. + :param limit_page_length: No of records in the page. Default 20. + :param group_by: Group by field. + :param or_filters: OR filter conditions. + :param debug: Debug mode. + :param as_dict: Return as dict format. + + Example usage: + - get_list("ToDo", fields=["name", "description"], filters={"owner":"test@example.com"}) + - get_list("ToDo", fields="*", filters=[["modified", ">", "2014-01-01"]]) + """ logger.info(f"API get_list: {pagetype}") - endpoint = f"api/data/{pagetype}" - data = self._make_request('GET', endpoint, kwargs) + # 构建查询参数 + params = {} + if fields is not None: + params['fields'] = fields + if filters is not None: + params['filters'] = filters + if group_by is not None: + params['group_by'] = group_by + if order_by is not None: + params['order_by'] = order_by + if limit_start is not None: + params['limit_start'] = limit_start + if limit_page_length != 20: # 默认值不需要传递 + params['limit_page_length'] = limit_page_length + if parent is not None: + params['parent'] = parent + if debug: + params['debug'] = debug + if not as_dict: # 默认 True，只有 False 时才传递 + params['as_dict'] = as_dict + if or_filters is not None: + params['or_filters'] = or_filters - # 转换为 Page 对象列表 - class MockPage(Page): - def __init__(self, pagetype, name=None, **kwargs): - d = {"pagetype": pagetype, **kwargs} - if name is not None: - d["name"] = name - super().__init__(d) + # 合并其他参数 + params.update(kwargs) - return [MockPage(pagetype, item.get('name'), **item) for item in data] + endpoint = self._build_endpoint(pagetype) + data = self._make_request('GET', endpoint, params) + + 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}") - - - class MockPage(Page): - def __init__(self, pagetype, name=None, **kwargs): - d = {"pagetype": pagetype, **kwargs} - if name is not None: - d["name"] = name - super().__init__(d) - - return MockPage(pagetype, None, **kwargs) + return self._create_page(pagetype, None, **kwargs) def save_pg(self, pg): """保存页面类型文档""" logger.info(f"API save_pg: {pg.pagetype}, {pg.name}") - endpoint = f"api/data/{pg.pagetype}" - if pg.name: - endpoint += f"/{pg.name}" - + endpoint = self._build_endpoint(pg.pagetype, pg.name) method = 'PUT' if pg.name else 'POST' data = pg.as_dict() @@ -137,16 +205,15 @@ class ApiAdapter: """删除页面类型文档""" logger.info(f"API delete_pg: {pagetype}, {name}") - endpoint = f"api/data/{pagetype}/{name}" + endpoint = self._build_endpoint(pagetype, name) self._make_request('DELETE', endpoint) - return True def db_exists(self, pagetype: str, filters: Dict[str, Any] = None): """检查文档是否存在""" logger.info(f"API db_exists: {pagetype}, {filters}") - endpoint = f"api/data/{pagetype}" + endpoint = self._build_endpoint(pagetype) params = {'filters': filters} if filters else {} data = self._make_request('GET', endpoint, params) @@ -156,7 +223,7 @@ class ApiAdapter: """获取数据库值""" logger.info(f"API db_get_value: {pagetype}, {filters}, {fieldname}") - endpoint = f"api/data/{pagetype}" + endpoint = self._build_endpoint(pagetype) params = {'filters': filters, 'fieldname': fieldname} if filters else {} data = self._make_request('GET', endpoint, params) @@ -166,7 +233,7 @@ class ApiAdapter: """设置数据库值""" logger.info(f"API db_set_value: {pagetype}, {filters}, {fieldname}, {value}") - endpoint = f"api/data/{pagetype}" + endpoint = self._build_endpoint(pagetype) data = {'filters': filters, 'fieldname': fieldname, 'value': value} self._make_request('PUT', endpoint, data)