get_pg和get_list对齐云端实现
This commit is contained in:
parent
48d35a5cdf
commit
040269478a
@ -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]] = {}
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user