Mobi Python SDK

List files in a folder using the Mobi Python SDK

# Import MobiClient to call the service
from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import PageFilesByFolderRequest
folder_id = "your_folder_id"
client = MobiClient()
# Call the page_files_by_folder service through MobiClient to list files in the specified folder
resp = client.page_files_by_folder(PageFilesByFolderRequest(folder_id=folder_id, page_size=500, page_number=1))
# Use get_data() to get data from the response class and use items to get the list of all files
if resp.success:
    files = resp.get_data()["items"]
    return files
else:
    return "Page files by folder failed, reason: " + resp.get_data()['message']

Result:

Delete a file by file ID using the Mobi Python SDK

# Import MobiClient to call the service
from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import DeleteFileRequest
# The ID of the file to delete
file_id = "your_file_id"
client = MobiClient()
# Call the delete_file interface to delete the 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']

Result:

Upload a file to the Mobi file system using the Mobi Python SDK

# Import MobiClient to call the service
import asyncio

from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import FileUploadRequest

# Generate a sample demo.txt file
with open("demo.txt", 'w', encoding='utf-8') as file:
    file.write('This is a demo file.\n')
    file.write('Welcome to the Python file operation feature!\n')
client = MobiClient()
# Construct the file upload request
file_request = FileUploadRequest(
        name="file_name_demo",           # File name: The name of the file after it is uploaded
        data=open("demo.txt", "rb"),     # File data: The file object opened in binary read-only mode
        folder_name="folder_name_demo",  # Optional folder name: The name of the destination folder. If the folder does not exist, it is created.
)
# Call the upload_file_to_mobi method to upload the file asynchronously
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']

Result:

Call a built-in database integration operation using the Mobi Python SDK

# Import MobiClient to call the service
from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import SystemPgConnector

client = MobiClient()
# Sample SQL
sql = "SELECT now();"
# Construct the integration operation request
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']

Result:

Call an HTTP integration operation using the Mobi Python SDK

Mobi HTTP resource configuration:

Resource ID:

# Import MobiClient to call the service
from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import HttpConnector

client = MobiClient()

# Construct the integration operation request
connector_request = HttpConnector(
        resource_id='175376942956017',
        url='/index/create',
        method='POST',
        header= {
                'x-acs-version': '2023-12-29',
                'x-acs-action': 'CreateIndex',
        },
        body={
            "Name": "new_knowledge_base_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']

Result:

Call an OpenAPI integration operation using the Mobi Python SDK

Mobi OpenAPI resource configuration:

Resource ID:

The OpenAPIConnector contains the following fields:

• resource_id: The ID of the Mobi integration resource.

• action: The action name of the OpenAPI request.

• method: The HTTP method in uppercase.

• path: The request URI. It can contain parameters in the {var1} format.

• path_params: A dictionary of key-value pairs for the parameters in the path.

• query_params: A dictionary of key-value pairs for the query parameters in the URL.

• body_params: A dictionary of key-value pairs for the body parameters.

• header_params: A dictionary of key-value pairs for the parameters in the HTTP request header.

The following example uses the Model Studio ListCategory operation:

1. action: ListCategory

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

3. method: POST

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

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

# Import MobiClient to call the service
from mobi_baas.client import MobiClient
# Import the request data model
from mobi_baas.models import OpenAPIConnector

client = MobiClient()

# Construct the integration operation request. Refer to the official documentation to fill in the action, path, and parameters for the corresponding locations.
# The following code constructs a Model Studio request as an example. For more information, see 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"},  # Path parameters, which replace parameters in the {var1} format in the path
    query_params= None, # Query parameters
    body_params= {"CategoryType": "UNSTRUCTURED"}, # Body parameters
    header_params= None # Header parameters
)
resp = client.invoke_connector(connector_request)
if resp.success:
    return resp.get_data()
else:
    return "invoke failed"+resp.get_data()['message']

Result: