Mobi Python SDK

通过 Mobi Python SDK 列出文件夹下的文件内容

# 引入 MobiClient 用于调用服务
from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import PageFilesByFolderRequest
folder_id = "your_folder_id"
client = MobiClient()
# 通过MobiCient调用 page_files_by_folder 服务列出指定文件夹下的文件
resp = client.page_files_by_folder(PageFilesByFolderRequest(folder_id=folder_id, page_size=500, page_number=1))
# 通过get_data()获取响应类中的数据，通过 items 获取所有文件列表
if resp.success:
    files = resp.get_data()["items"]
    return files
else:
    return "Page files by folder failed, reason: " + resp.get_data()['message']

运行结果：

通过 Mobi Python SDK 根据文件 ID 删除文件

# 引入 MobiClient 用于调用服务
from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import DeleteFileRequest
# 要删除的文件 ID
file_id = "your_file_id"
client = MobiClient()
# 调用 delete_file 接口删除文件
resp = client.delete_file(DeleteFileRequest(file_id=file_id))
if resp.success:
    return "File deleted"
else:
    return "File delete failed, reason: " + resp.get_data()['message']

运行结果：

通过 Mobi Python SDK 上传文件到魔笔的文件系统

# 引入 MobiClient 用于调用服务
import asyncio

from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import FileUploadRequest

# 生成示例的demo.txt文件
with open("demo.txt", 'w', encoding='utf-8') as file:
    file.write('这是一个demo示例文件。\n')
    file.write('欢迎使用Python文件操作功能！\n')
client = MobiClient()
# 构造上传文件请求
file_request = FileUploadRequest(
        name="file_name_demo",           # 文件名：上传后的文件名称
        data=open("demo.txt", "rb"),     # 文件数据：以二进制只读模式打开的文件对象
        folder_name="folder_name_demo",  # 可选文件夹名称：指定文件上传到的目标文件夹名称，不存在会创建
)
# 调用upload_file_to_mobi方法异步上传文件
result = asyncio.run(client.upload_file_to_mobi(file_request))
file_id = ""
if result.success:
    file_id = result.get_data().get("fileId", "")
    return file_id
else:
    return "File upload failed"+ result.get_data()['message']

运行结果：

通过 Mobi Python SDK 调用内置数据库集成操作

# 引入 MobiClient 用于调用服务
from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import SystemPgConnector

client = MobiClient()
# 示例sql
sql = "SELECT now();"
# 构造集成操作请求
connector_request = SystemPgConnector(
    sql=sql
)
resp = client.invoke_connector(connector_request)
if resp.success:
    return resp.get_data()
else:
    return "invoke failed"+resp.get_data()['message']

运行结果：

通过 Mobi Python SDK 调用HTTP集成操作

魔笔HTTP资源配置：

资源ID:

# 引入 MobiClient 用于调用服务
from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import HttpConnector

client = MobiClient()

# 构造集成操作请求
connector_request = HttpConnector(
        resource_id='175376942956017',
        url='/index/create',
        method='POST',
        header= {
                'x-acs-version': '2023-12-29',
                'x-acs-action': 'CreateIndex',
        },
        body={
            "Name": "新建知识库demo",
            "StructureType": "unstructured",
            "SinkType": "BUILT_IN"
        }
)
resp = client.invoke_connector(connector_request)
if resp.success:
    return resp.get_data()
else:
    return "invoke failed"+resp.get_data()['message']

运行结果：

通过 Mobi Python SDK 调用 OpenAPI 集成操作

魔笔 OpenAPI 资源配置：

资源ID:

对于OpenAPIConnnector，其中包含的字段如下：

• resource_id: 魔笔集成资源的 id

• action: OpenAPI 请求的操作名称

• method: 大写的 HTTP 方法

• path: 请求路径，可以包含参数，参数使用{var1}的格式

• path_params: Dict 类型，对应填写 path 中参数的键值对

• query_params: Dict 类型，url 中的查询参数的键值对

• body_params: Dict 类型， body 参数键值对

• header_params: Dict 类型，http 请求 header 中的参数键值对

下面以百炼 ListCategory为例：

1. action: ListCategory

2. path: /{WorkspaceId}/datacenter/categories

3. methd: POST

4. path_params: {"WorkspaceId": "bailianWorkspaceId"}

5. body_params: {"CategoryType": "UNSTRUCTURED"}

# 引入 MobiClient 用于调用服务
from mobi_baas.client import MobiClient
# 引入请求数据模型
from mobi_baas.models import OpenAPIConnector

client = MobiClient()

# 构造集成操作请求，需要参照官方文档，填写action和path，以及对应位置的参数
# 下面构造一个百炼请求作为示例，文档可以参考https://help.aliyun.com/zh/model-studio/api-bailian-2023-12-29-listcategory
connector_request = OpenAPIConnector(
    resource_id='174305288320562',
    action= "ListCategory",
    method="POST",
    path = "/{WorkspaceId}/datacenter/categories",
    path_params= {"WorkspaceId": "bailianWorkspaceId"},  #路径参数，替换路径中的{var1}格式的参数
    query_params= None, #查询参数
    body_params= {"CategoryType": "UNSTRUCTURED"}, #body参数
    header_params= None #header参数
)
resp = client.invoke_connector(connector_request)
if resp.success:
    return resp.get_data()
else:
    return "invoke failed"+resp.get_data()['message']

运行结果：