adai/MemoryBear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(app): add cross-workspace app sharing with auto-rename on import

Browse Source
This commit is contained in:
wxy
2026-03-16 16:16:02 +08:00
parent 81a232177e
commit 0832337839
5 changed files with 535 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
api/app/services/app_dsl_service.py
View File

@@ -11,6 +11,7 @@ from app.core.error_codes import BizCode
from app.core.exceptions import BusinessException, ResourceNotFoundException
from app.models import AgentConfig, MultiAgentConfig
from app.models.app_model import App, AppType
from app.models.appshare_model import AppShare
from app.models.app_release_model import AppRelease
from app.models.knowledge_model import Knowledge
from app.models.models_model import ModelConfig
@@ -220,7 +221,7 @@ class AppDslService:
id=uuid.uuid4(),
workspace_id=workspace_id,
created_by=user_id,
name=app_meta.get("name", "导入应用"),
name=self._unique_app_name(app_meta.get("name", "导入应用"), workspace_id, app_type),
description=app_meta.get("description"),
icon=app_meta.get("icon"),
icon_type=app_meta.get("icon_type"),
@@ -296,6 +297,30 @@ class AppDslService:
self.db.refresh(new_app)
return new_app, warnings
def _unique_app_name(self, name: str, workspace_id: uuid.UUID, app_type: AppType) -> str:
"""生成唯一应用名称，同时检查本空间自有应用和共享到本空间的应用"""
# 本空间自有应用名
existing = {r[0] for r in self.db.query(App.name).filter(
App.workspace_id == workspace_id,
App.type == app_type,
App.is_active.is_(True)
).all()}
# 共享到本空间的应用名
shared_names = {r[0] for r in self.db.query(App.name).join(
AppShare, AppShare.source_app_id == App.id
).filter(
AppShare.target_workspace_id == workspace_id,
App.type == app_type,
App.is_active.is_(True)
).all()}
existing |= shared_names
if name not in existing:
return name
counter = 1
while f"{name}({counter})" in existing:
counter += 1
return f"{name}({counter})"
def _resolve_model(self, ref: Optional[dict], tenant_id: uuid.UUID, warnings: list) -> Optional[uuid.UUID]:
if not ref:
return None

503
api/app/services/app_service.py
View File

@@ -7,12 +7,13 @@
- 应用发布和版本管理
- 应用回滚
"""
import copy
import datetime
import uuid
from typing import Annotated, Any, Dict, List, Optional, Tuple
from fastapi import Depends
from sqlalchemy import and_, func, or_, select
from sqlalchemy import and_, delete, func, or_, select
from sqlalchemy.orm import Session
from app.core.error_codes import BizCode
@@ -80,6 +81,8 @@ class AppService:
)
raise BusinessException("应用不在指定工作空间中", BizCode.WORKSPACE_NO_ACCESS)
def _check_app_accessible(self, app: App, workspace_id: Optional[uuid.UUID]) -> bool:
"""检查应用是否可访问（包括共享应用）
@@ -102,7 +105,8 @@ class AppService:
# 2. 检查是否是共享给本工作空间的应用
stmt = select(AppShare).where(
AppShare.source_app_id == app.id,
AppShare.target_workspace_id == workspace_id
AppShare.target_workspace_id == workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
@@ -140,17 +144,18 @@ class AppService:
stmt = select(AppShare).where(
AppShare.source_app_id == app.id,
AppShare.target_workspace_id == workspace_id
AppShare.target_workspace_id == workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
return share.permission if share else None
def _validate_app_writable(self, app: App, workspace_id: Optional[uuid.UUID]) -> None:
"""Validate that the app config is writable (owner only).
"""Validate that the app config is writable.
Shared apps (both readonly and editable) cannot modify config.
- Own workspace app: allowed
- Any shared app: denied
- Shared app with editable permission: allowed
- Shared app with readonly permission: denied
Raises:
BusinessException: when app is not writable
@@ -162,6 +167,11 @@ class AppService:
if app.workspace_id == workspace_id:
return
# Check share permission
permission = self._get_share_permission(app, workspace_id)
if permission == "editable":
return
logger.warning(
"应用写操作被拒",
extra={"app_id": str(app.id), "workspace_id": str(workspace_id)}
@@ -503,17 +513,27 @@ class AppService:
source_workspace_icon = None
source_app_version = None
source_app_is_active = None
share_id = None
shared_by = None
shared_by_name = None
shared_at = None
if is_shared:
# 查询共享权限和来源工作空间名称
from app.models import AppShare
stmt = select(AppShare).where(
AppShare.source_app_id == app.id,
AppShare.target_workspace_id == current_workspace_id
AppShare.target_workspace_id == current_workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
if share:
share_id = share.id
share_permission = share.permission
shared_by = share.shared_by
shared_at = share.created_at
if share.shared_user:
shared_by_name = share.shared_user.username
if share.source_workspace:
source_workspace_name = share.source_workspace.name
source_workspace_icon = share.source_workspace.icon
@@ -543,6 +563,10 @@ class AppService:
"source_workspace_icon": source_workspace_icon,
"source_app_version": source_app_version,
"source_app_is_active": source_app_is_active,
"share_id": share_id,
"shared_by": shared_by,
"shared_by_name": shared_by_name,
"shared_at": shared_at,
"created_at": app.created_at,
"updated_at": app.updated_at
}
@@ -780,6 +804,19 @@ class AppService:
self.db.add(new_app)
self.db.flush()
# 判断是否跨工作空间复制（共享应用复制到自己的工作空间）
is_cross_workspace = target_workspace_id != source_app.workspace_id
# 跨工作空间时，获取目标工作空间的 tenant_id 用于判断模型配置是否可用
target_tenant_id = None
available_model_ids: set = set()
available_kb_ids: set = set()
if is_cross_workspace:
target_ws = self.db.get(Workspace, target_workspace_id)
if not target_ws:
raise ResourceNotFoundException("工作空间", str(target_workspace_id))
target_tenant_id = target_ws.tenant_id
# 如果是 agent 类型，复制 AgentConfig
if source_app.type == AppType.AGENT:
source_config = self.db.query(AgentConfig).filter(
@@ -787,16 +824,40 @@ class AppService:
).first()
if source_config:
if is_cross_workspace:
# Batch-collect and preload all referenced resources
model_ids, kb_ids = self._collect_resource_ids_from_config(
source_config.default_model_config_id,
source_config.knowledge_retrieval,
source_config.tools
)
available_model_ids, available_kb_ids = self._preload_cross_workspace_resources(
target_tenant_id, target_workspace_id, model_ids, kb_ids
)
new_model_config_id = self._is_model_available(
source_config.default_model_config_id, available_model_ids
)
new_knowledge_retrieval = self._clean_knowledge_retrieval(
source_config.knowledge_retrieval, available_kb_ids
)
new_tools = self._clean_tools(
source_config.tools, available_kb_ids
)
else:
new_model_config_id = source_config.default_model_config_id
new_knowledge_retrieval = copy.deepcopy(source_config.knowledge_retrieval) if source_config.knowledge_retrieval else None
new_tools = copy.deepcopy(source_config.tools) if source_config.tools else []
new_config = AgentConfig(
id=uuid.uuid4(),
app_id=new_app.id,
system_prompt=source_config.system_prompt,
default_model_config_id=source_config.default_model_config_id,
model_parameters=source_config.model_parameters.copy() if source_config.model_parameters else None,
knowledge_retrieval=source_config.knowledge_retrieval.copy() if source_config.knowledge_retrieval else None,
memory=source_config.memory.copy() if source_config.memory else None,
variables=source_config.variables.copy() if source_config.variables else [],
tools=source_config.tools.copy() if source_config.tools else [],
default_model_config_id=new_model_config_id,
model_parameters=copy.deepcopy(source_config.model_parameters) if source_config.model_parameters else None,
knowledge_retrieval=new_knowledge_retrieval,
memory=copy.deepcopy(source_config.memory) if source_config.memory else None,
variables=copy.deepcopy(source_config.variables) if source_config.variables else [],
tools=new_tools,
is_active=True,
created_at=now,
updated_at=now,
@@ -809,14 +870,29 @@ class AppService:
).first()
if source_config:
if is_cross_workspace:
model_ids, kb_ids = self._collect_resource_ids_from_workflow_nodes(
source_config.nodes
)
available_model_ids, available_kb_ids = self._preload_cross_workspace_resources(
target_tenant_id, target_workspace_id, model_ids, kb_ids
)
new_nodes = self._clean_workflow_nodes_for_cross_workspace(
source_config.nodes or [],
available_model_ids,
available_kb_ids
)
else:
new_nodes = copy.deepcopy(source_config.nodes) if source_config.nodes else []
new_config = WorkflowConfig(
id=uuid.uuid4(),
app_id=new_app.id,
nodes=source_config.nodes.copy() if source_config.nodes else [],
edges=source_config.edges.copy() if source_config.edges else [],
variables=source_config.variables.copy() if source_config.variables else [],
execution_config=source_config.execution_config.copy() if source_config.execution_config else {},
triggers=source_config.triggers.copy() if source_config.triggers else [],
nodes=new_nodes,
edges=copy.deepcopy(source_config.edges) if source_config.edges else [],
variables=copy.deepcopy(source_config.variables) if source_config.variables else [],
execution_config=copy.deepcopy(source_config.execution_config) if source_config.execution_config else {},
triggers=copy.deepcopy(source_config.triggers) if source_config.triggers else [],
is_active=True,
created_at=now,
updated_at=now,
@@ -829,17 +905,28 @@ class AppService:
).first()
if source_config:
if is_cross_workspace:
model_ids = {source_config.default_model_config_id} if source_config.default_model_config_id else set()
available_model_ids, _ = self._preload_cross_workspace_resources(
target_tenant_id, target_workspace_id, model_ids, set()
)
new_model_config_id = self._is_model_available(
source_config.default_model_config_id, available_model_ids
)
else:
new_model_config_id = source_config.default_model_config_id
new_config = MultiAgentConfig(
id=uuid.uuid4(),
app_id=new_app.id,
master_agent_id=source_config.master_agent_id,
master_agent_id=source_config.master_agent_id if not is_cross_workspace else None,
master_agent_name=source_config.master_agent_name,
default_model_config_id=source_config.default_model_config_id,
default_model_config_id=new_model_config_id,
model_parameters=source_config.model_parameters,
orchestration_mode=source_config.orchestration_mode,
sub_agents=source_config.sub_agents.copy() if source_config.sub_agents else [],
routing_rules=source_config.routing_rules.copy() if source_config.routing_rules else None,
execution_config=source_config.execution_config.copy() if source_config.execution_config else {},
sub_agents=copy.deepcopy(source_config.sub_agents) if source_config.sub_agents else [],
routing_rules=copy.deepcopy(source_config.routing_rules) if source_config.routing_rules else None,
execution_config=copy.deepcopy(source_config.execution_config) if source_config.execution_config else {},
aggregation_strategy=source_config.aggregation_strategy,
is_active=True,
created_at=now,
@@ -869,6 +956,241 @@ class AppService:
)
raise BusinessException(f"应用复制失败: {str(e)}", BizCode.INTERNAL_ERROR, cause=e)
def _preload_cross_workspace_resources(
self,
target_tenant_id: Optional[uuid.UUID],
target_workspace_id: uuid.UUID,
model_config_ids: set,
kb_ids: set
) -> tuple:
"""Batch-load model configs and knowledge bases to avoid N+1 queries.
Returns:
(available_model_ids, available_kb_ids): sets of IDs available in target workspace
"""
from app.models.models_model import ModelConfig as MC
from app.models.knowledge_model import Knowledge
from app.models.knowledgeshare_model import KnowledgeShare
# Batch check model configs by tenant
available_model_ids: set = set()
if model_config_ids and target_tenant_id:
stmt = select(MC.id).where(
MC.id.in_(model_config_ids),
MC.tenant_id == target_tenant_id
)
available_model_ids = set(self.db.scalars(stmt).all())
# Batch check knowledge bases
available_kb_ids: set = set()
if kb_ids:
kb_uuids = set()
for kid in kb_ids:
try:
kb_uuids.add(uuid.UUID(str(kid)))
except (ValueError, AttributeError):
pass
if kb_uuids:
# KBs in target workspace
stmt = select(Knowledge.id).where(
Knowledge.id.in_(kb_uuids),
Knowledge.workspace_id == target_workspace_id
)
available_kb_ids.update(self.db.scalars(stmt).all())
# KBs shared to target workspace
remaining = kb_uuids - available_kb_ids
if remaining:
stmt = select(KnowledgeShare.source_kb_id).where(
KnowledgeShare.source_kb_id.in_(remaining),
KnowledgeShare.target_workspace_id == target_workspace_id
)
available_kb_ids.update(self.db.scalars(stmt).all())
return available_model_ids, available_kb_ids
@staticmethod
def _collect_resource_ids_from_config(
model_config_id: Optional[uuid.UUID],
knowledge_retrieval: Optional[dict],
tools: Optional[list]
) -> tuple:
"""Extract all model config IDs and knowledge base IDs from an app config."""
model_ids: set = set()
kb_ids: set = set()
if model_config_id:
model_ids.add(model_config_id)
if knowledge_retrieval and isinstance(knowledge_retrieval, dict):
if "kb_ids" in knowledge_retrieval:
for kid in knowledge_retrieval.get("kb_ids", []):
if kid:
kb_ids.add(str(kid))
if knowledge_retrieval.get("knowledge_id"):
kb_ids.add(str(knowledge_retrieval["knowledge_id"]))
if tools:
for tool in tools:
if isinstance(tool, dict):
kid = tool.get("knowledge_id") or tool.get("kb_id")
if kid:
kb_ids.add(str(kid))
return model_ids, kb_ids
@staticmethod
def _collect_resource_ids_from_workflow_nodes(nodes: list) -> tuple:
"""Extract all model config IDs and knowledge base IDs from workflow nodes."""
model_ids: set = set()
kb_ids: set = set()
for node in (nodes or []):
if not isinstance(node, dict):
continue
data = node.get("data", {})
if not isinstance(data, dict):
continue
for key in ("model_config_id", "default_model_config_id"):
val = data.get(key)
if val:
try:
model_ids.add(uuid.UUID(str(val)))
except (ValueError, AttributeError):
pass
kr = data.get("knowledge_retrieval")
if isinstance(kr, dict):
for kid in kr.get("kb_ids", []):
if kid:
kb_ids.add(str(kid))
if kr.get("knowledge_id"):
kb_ids.add(str(kr["knowledge_id"]))
if data.get("knowledge_id"):
kb_ids.add(str(data["knowledge_id"]))
for kid in data.get("kb_ids", []):
if kid:
kb_ids.add(str(kid))
return model_ids, kb_ids
@staticmethod
def _is_model_available(model_config_id: Optional[uuid.UUID], available_model_ids: set) -> Optional[uuid.UUID]:
if not model_config_id:
return None
return model_config_id if model_config_id in available_model_ids else None
@staticmethod
def _is_kb_available(kb_id: Optional[str], available_kb_ids: set) -> Optional[str]:
if not kb_id:
return None
try:
return kb_id if uuid.UUID(str(kb_id)) in available_kb_ids else None
except (ValueError, AttributeError):
return None
def _clean_knowledge_retrieval(
self,
knowledge_retrieval: Optional[dict],
available_kb_ids: set
) -> Optional[dict]:
"""Clean knowledge retrieval config, keeping only available KBs."""
if not knowledge_retrieval:
return None
cleaned = copy.deepcopy(knowledge_retrieval)
if "kb_ids" in cleaned and isinstance(cleaned["kb_ids"], list):
cleaned["kb_ids"] = [
kid for kid in cleaned["kb_ids"]
if self._is_kb_available(kid, available_kb_ids)
]
if "knowledge_id" in cleaned:
cleaned["knowledge_id"] = self._is_kb_available(
cleaned.get("knowledge_id"), available_kb_ids
)
return cleaned
def _clean_tools(
self,
tools: Optional[list],
available_kb_ids: set
) -> list:
"""Clean tools config, keeping built-in tools and tools with available KBs."""
if not tools:
return []
cleaned = []
for tool in tools:
if not isinstance(tool, dict):
cleaned.append(tool)
continue
tool_type = tool.get("type", "")
if tool_type in ("builtin", "built_in", "system"):
cleaned.append(copy.deepcopy(tool))
continue
kb_id = tool.get("knowledge_id") or tool.get("kb_id")
if kb_id:
if self._is_kb_available(kb_id, available_kb_ids):
cleaned.append(copy.deepcopy(tool))
continue
cleaned.append(copy.deepcopy(tool))
return cleaned
def _clean_workflow_nodes_for_cross_workspace(
self,
nodes: list,
available_model_ids: set,
available_kb_ids: set
) -> list:
"""Clean workflow nodes, using pre-loaded resource sets. Uses deepcopy to avoid mutating source."""
if not nodes:
return []
cleaned = []
for node in nodes:
if not isinstance(node, dict):
cleaned.append(node)
continue
node_copy = copy.deepcopy(node)
data = node_copy.get("data")
if not isinstance(data, dict):
cleaned.append(node_copy)
continue
for key in ("model_config_id", "default_model_config_id"):
if key in data and data[key]:
try:
mid = uuid.UUID(str(data[key]))
except (ValueError, AttributeError):
data[key] = None
continue
data[key] = str(mid) if mid in available_model_ids else None
if "knowledge_retrieval" in data and data["knowledge_retrieval"]:
data["knowledge_retrieval"] = self._clean_knowledge_retrieval(
data["knowledge_retrieval"], available_kb_ids
)
if "knowledge_id" in data:
data["knowledge_id"] = self._is_kb_available(
data.get("knowledge_id"), available_kb_ids
)
if "kb_ids" in data and isinstance(data["kb_ids"], list):
data["kb_ids"] = [
kid for kid in data["kb_ids"]
if self._is_kb_available(kid, available_kb_ids)
]
cleaned.append(node_copy)
return cleaned
def list_apps(
self,
*,
@@ -878,6 +1200,7 @@ class AppService:
status: Optional[str] = None,
search: Optional[str] = None,
include_shared: bool = True,
shared_only: bool = False,
page: int = 1,
pagesize: int = 10,
) -> Tuple[List[App], int]:
@@ -923,18 +1246,24 @@ class AppService:
if search:
filters.append(func.lower(App.name).like(f"%{search.lower()}%"))
# 基础查询：本工作空间的应用
if include_shared:
# 查询本工作空间的应用 + 分享给本工作空间的应用
# 使用 OR 条件：workspace_id = current OR app_id IN (shared apps)
# shared_only implies include_shared; enforce to avoid confusing API usage
if shared_only:
include_shared = True
# 获取分享给本工作空间的应用ID列表
# 基础查询：本工作空间的应用
if shared_only:
# 只返回共享给本工作空间的应用，不含自有应用
shared_app_ids_stmt = (
select(AppShare.source_app_id)
.where(AppShare.target_workspace_id == workspace_id)
.where(AppShare.target_workspace_id == workspace_id, AppShare.is_active.is_(True))
)
stmt = select(App).where(App.id.in_(shared_app_ids_stmt))
elif include_shared:
# 查询本工作空间的应用 + 分享给本工作空间的应用
shared_app_ids_stmt = (
select(AppShare.source_app_id)
.where(AppShare.target_workspace_id == workspace_id, AppShare.is_active.is_(True))
)
# 构建主查询：本工作空间的应用 OR 分享的应用
stmt = select(App).where(
or_(
App.workspace_id == workspace_id,
@@ -1789,7 +2118,8 @@ class AppService:
# 检查是否已经分享过
stmt = select(AppShare).where(
AppShare.source_app_id == app_id,
AppShare.target_workspace_id == target_ws_id
AppShare.target_workspace_id == target_ws_id,
AppShare.is_active.is_(True)
)
existing_share = self.db.scalars(stmt).first()
@@ -1868,7 +2198,8 @@ class AppService:
# 2. 查找分享记录
stmt = select(AppShare).where(
AppShare.source_app_id == app_id,
AppShare.target_workspace_id == target_workspace_id
AppShare.target_workspace_id == target_workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
@@ -1882,8 +2213,8 @@ class AppService:
f"app_id={app_id}, target_workspace_id={target_workspace_id}"
)
# 3. 删除分享记录
self.db.delete(share)
# 3. 逻辑删除分享记录
share.is_active = False
self.db.commit()
logger.info(
@@ -1891,6 +2222,48 @@ class AppService:
extra={"app_id": str(app_id), "target_workspace_id": str(target_workspace_id)}
)
def unshare_all_apps_to_workspace(
self,
*,
target_workspace_id: uuid.UUID,
workspace_id: uuid.UUID
) -> int:
"""Cancel all app shares from current workspace to a target workspace.
Args:
target_workspace_id: Target workspace ID to cancel all shares to
workspace_id: Current workspace ID (source)
Returns:
Number of share records deleted
"""
from app.models import AppShare
logger.info(
"取消对目标工作空间的所有应用分享",
extra={"target_workspace_id": str(target_workspace_id), "workspace_id": str(workspace_id)}
)
# Query active records first for reliable count
id_stmt = select(AppShare.id).where(
AppShare.source_workspace_id == workspace_id,
AppShare.target_workspace_id == target_workspace_id,
AppShare.is_active.is_(True)
)
ids = list(self.db.scalars(id_stmt).all())
count = len(ids)
if ids:
# Soft delete: mark as inactive
from sqlalchemy import update as sa_update
self.db.execute(
sa_update(AppShare).where(AppShare.id.in_(ids)).values(is_active=False)
)
self.db.commit()
logger.info("已取消分享记录数", extra={"count": count})
return count
def list_app_shares(
self,
*,
@@ -1920,7 +2293,8 @@ class AppService:
# 查询分享记录
stmt = select(AppShare).where(
AppShare.source_app_id == app_id
AppShare.source_app_id == app_id,
AppShare.is_active.is_(True)
).order_by(AppShare.created_at.desc())
shares = list(self.db.scalars(stmt).all())
@@ -1958,7 +2332,8 @@ class AppService:
stmt = select(AppShare).where(
AppShare.source_app_id == app_id,
AppShare.target_workspace_id == workspace_id
AppShare.target_workspace_id == workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
@@ -1968,7 +2343,8 @@ class AppService:
f"app_id={app_id}, workspace_id={workspace_id}"
)
self.db.delete(share)
# Soft delete
share.is_active = False
self.db.commit()
logger.info(
@@ -1976,6 +2352,47 @@ class AppService:
extra={"app_id": str(app_id), "workspace_id": str(workspace_id)}
)
def remove_all_shared_apps_from_workspace(
self,
*,
source_workspace_id: uuid.UUID,
workspace_id: uuid.UUID
) -> int:
"""Remove all shared apps from a specific source workspace.
Args:
source_workspace_id: The workspace that shared the apps
workspace_id: Current workspace ID (recipient)
Returns:
Number of share records deleted
"""
from app.models import AppShare
logger.info(
"批量移除来源工作空间的共享应用",
extra={"source_workspace_id": str(source_workspace_id), "workspace_id": str(workspace_id)}
)
# Query active records for reliable count, then soft delete
id_stmt = select(AppShare.id).where(
AppShare.source_workspace_id == source_workspace_id,
AppShare.target_workspace_id == workspace_id,
AppShare.is_active.is_(True)
)
ids = list(self.db.scalars(id_stmt).all())
count = len(ids)
if ids:
from sqlalchemy import update as sa_update
self.db.execute(
sa_update(AppShare).where(AppShare.id.in_(ids)).values(is_active=False)
)
self.db.commit()
logger.info("已移除共享记录数", extra={"count": count})
return count
def list_my_shared_out(
self,
*,
@@ -1990,7 +2407,10 @@ class AppService:
stmt = (
select(AppShare)
.where(AppShare.source_workspace_id == workspace_id)
.where(
AppShare.source_workspace_id == workspace_id,
AppShare.is_active.is_(True)
)
.order_by(AppShare.created_at.desc())
)
return list(self.db.scalars(stmt).all())
@@ -2023,7 +2443,8 @@ class AppService:
stmt = select(AppShare).where(
AppShare.source_app_id == app_id,
AppShare.target_workspace_id == target_workspace_id
AppShare.target_workspace_id == target_workspace_id,
AppShare.is_active.is_(True)
)
share = self.db.scalars(stmt).first()
@@ -2139,6 +2560,7 @@ def list_apps(
status: Optional[str] = None,
search: Optional[str] = None,
include_shared: bool = True,
shared_only: bool = False,
page: int = 1,
pagesize: int = 10,
) -> Tuple[List[App], int]:
@@ -2151,6 +2573,7 @@ def list_apps(
status=status,
search=search,
include_shared=include_shared,
shared_only=shared_only,
page=page,
pagesize=pagesize,
)