sdk/dingdingSdk/alibabacloud_dingtalk/sns_storage_1_0/client.py

1312 lines
53 KiB
Python
Raw Permalink Normal View History

2025-08-19 10:20:23 +00:00
# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
from Tea.core import TeaCore
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_gateway_dingtalk.client import Client as GatewayClientClient
from alibabacloud_tea_util.client import Client as UtilClient
from alibabacloud_dingtalk.sns_storage_1_0 import models as dingtalksns_storage__1__0_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
class Client(OpenApiClient):
"""
*\
"""
def __init__(
self,
config: open_api_models.Config,
):
super().__init__(config)
gateway_client = GatewayClientClient()
self._spi = gateway_client
self._endpoint_rule = ''
if UtilClient.empty(self._endpoint):
self._endpoint = 'api.dingtalk.com'
def get_dentries_with_options(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentriesRequest,
headers: dingtalksns_storage__1__0_models.GetDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentriesResponse:
"""
@summary 三方个人应用批量获取文件或文件夹信息
@param request: GetDentriesRequest
@param headers: GetDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.dentry_ids):
body['dentryIds'] = request.dentry_ids
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/batchQuery',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentriesResponse(),
self.execute(params, req, runtime)
)
async def get_dentries_with_options_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentriesRequest,
headers: dingtalksns_storage__1__0_models.GetDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentriesResponse:
"""
@summary 三方个人应用批量获取文件或文件夹信息
@param request: GetDentriesRequest
@param headers: GetDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.dentry_ids):
body['dentryIds'] = request.dentry_ids
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/batchQuery',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentriesResponse(),
await self.execute_async(params, req, runtime)
)
def get_dentries(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentriesRequest,
) -> dingtalksns_storage__1__0_models.GetDentriesResponse:
"""
@summary 三方个人应用批量获取文件或文件夹信息
@param request: GetDentriesRequest
@return: GetDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentriesHeaders()
return self.get_dentries_with_options(space_id, request, headers, runtime)
async def get_dentries_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentriesRequest,
) -> dingtalksns_storage__1__0_models.GetDentriesResponse:
"""
@summary 三方个人应用批量获取文件或文件夹信息
@param request: GetDentriesRequest
@return: GetDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentriesHeaders()
return await self.get_dentries_with_options_async(space_id, request, headers, runtime)
def get_dentry_with_options(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetDentryRequest,
headers: dingtalksns_storage__1__0_models.GetDentryHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentryResponse:
"""
@summary 三方个人应用获取文件()信息
@param request: GetDentryRequest
@param headers: GetDentryHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentryResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentry',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/{dentry_id}/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentryResponse(),
self.execute(params, req, runtime)
)
async def get_dentry_with_options_async(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetDentryRequest,
headers: dingtalksns_storage__1__0_models.GetDentryHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentryResponse:
"""
@summary 三方个人应用获取文件()信息
@param request: GetDentryRequest
@param headers: GetDentryHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentryResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentry',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/{dentry_id}/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentryResponse(),
await self.execute_async(params, req, runtime)
)
def get_dentry(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetDentryRequest,
) -> dingtalksns_storage__1__0_models.GetDentryResponse:
"""
@summary 三方个人应用获取文件()信息
@param request: GetDentryRequest
@return: GetDentryResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentryHeaders()
return self.get_dentry_with_options(space_id, dentry_id, request, headers, runtime)
async def get_dentry_async(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetDentryRequest,
) -> dingtalksns_storage__1__0_models.GetDentryResponse:
"""
@summary 三方个人应用获取文件()信息
@param request: GetDentryRequest
@return: GetDentryResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentryHeaders()
return await self.get_dentry_with_options_async(space_id, dentry_id, request, headers, runtime)
def get_dentry_thumbnails_with_options(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentryThumbnailsRequest,
headers: dingtalksns_storage__1__0_models.GetDentryThumbnailsHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse:
"""
@summary 三方个人应用批量获取文件缩略图
@param request: GetDentryThumbnailsRequest
@param headers: GetDentryThumbnailsHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentryThumbnailsResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.dentry_ids):
body['dentryIds'] = request.dentry_ids
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentryThumbnails',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/thumbnails/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse(),
self.execute(params, req, runtime)
)
async def get_dentry_thumbnails_with_options_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentryThumbnailsRequest,
headers: dingtalksns_storage__1__0_models.GetDentryThumbnailsHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse:
"""
@summary 三方个人应用批量获取文件缩略图
@param request: GetDentryThumbnailsRequest
@param headers: GetDentryThumbnailsHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetDentryThumbnailsResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.dentry_ids):
body['dentryIds'] = request.dentry_ids
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetDentryThumbnails',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/thumbnails/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse(),
await self.execute_async(params, req, runtime)
)
def get_dentry_thumbnails(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentryThumbnailsRequest,
) -> dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse:
"""
@summary 三方个人应用批量获取文件缩略图
@param request: GetDentryThumbnailsRequest
@return: GetDentryThumbnailsResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentryThumbnailsHeaders()
return self.get_dentry_thumbnails_with_options(space_id, request, headers, runtime)
async def get_dentry_thumbnails_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.GetDentryThumbnailsRequest,
) -> dingtalksns_storage__1__0_models.GetDentryThumbnailsResponse:
"""
@summary 三方个人应用批量获取文件缩略图
@param request: GetDentryThumbnailsRequest
@return: GetDentryThumbnailsResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetDentryThumbnailsHeaders()
return await self.get_dentry_thumbnails_with_options_async(space_id, request, headers, runtime)
def get_file_download_info_with_options(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetFileDownloadInfoRequest,
headers: dingtalksns_storage__1__0_models.GetFileDownloadInfoHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse:
"""
@summary 三方个人应用获取文件下载信息
@param request: GetFileDownloadInfoRequest
@param headers: GetFileDownloadInfoHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetFileDownloadInfoResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetFileDownloadInfo',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/{dentry_id}/downloadInfos/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse(),
self.execute(params, req, runtime)
)
async def get_file_download_info_with_options_async(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetFileDownloadInfoRequest,
headers: dingtalksns_storage__1__0_models.GetFileDownloadInfoHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse:
"""
@summary 三方个人应用获取文件下载信息
@param request: GetFileDownloadInfoRequest
@param headers: GetFileDownloadInfoHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetFileDownloadInfoResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetFileDownloadInfo',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/{dentry_id}/downloadInfos/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse(),
await self.execute_async(params, req, runtime)
)
def get_file_download_info(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetFileDownloadInfoRequest,
) -> dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse:
"""
@summary 三方个人应用获取文件下载信息
@param request: GetFileDownloadInfoRequest
@return: GetFileDownloadInfoResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetFileDownloadInfoHeaders()
return self.get_file_download_info_with_options(space_id, dentry_id, request, headers, runtime)
async def get_file_download_info_async(
self,
space_id: str,
dentry_id: str,
request: dingtalksns_storage__1__0_models.GetFileDownloadInfoRequest,
) -> dingtalksns_storage__1__0_models.GetFileDownloadInfoResponse:
"""
@summary 三方个人应用获取文件下载信息
@param request: GetFileDownloadInfoRequest
@return: GetFileDownloadInfoResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetFileDownloadInfoHeaders()
return await self.get_file_download_info_with_options_async(space_id, dentry_id, request, headers, runtime)
def get_space_with_options(
self,
request: dingtalksns_storage__1__0_models.GetSpaceRequest,
headers: dingtalksns_storage__1__0_models.GetSpaceHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetSpaceResponse:
"""
@summary 三方个人应用获取IM会话存储空间信息
@param request: GetSpaceRequest
@param headers: GetSpaceHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetSpaceResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.open_conversation_id):
body['openConversationId'] = request.open_conversation_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetSpace',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/conversations/spaces/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetSpaceResponse(),
self.execute(params, req, runtime)
)
async def get_space_with_options_async(
self,
request: dingtalksns_storage__1__0_models.GetSpaceRequest,
headers: dingtalksns_storage__1__0_models.GetSpaceHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.GetSpaceResponse:
"""
@summary 三方个人应用获取IM会话存储空间信息
@param request: GetSpaceRequest
@param headers: GetSpaceHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: GetSpaceResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.open_conversation_id):
body['openConversationId'] = request.open_conversation_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='GetSpace',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/conversations/spaces/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.GetSpaceResponse(),
await self.execute_async(params, req, runtime)
)
def get_space(
self,
request: dingtalksns_storage__1__0_models.GetSpaceRequest,
) -> dingtalksns_storage__1__0_models.GetSpaceResponse:
"""
@summary 三方个人应用获取IM会话存储空间信息
@param request: GetSpaceRequest
@return: GetSpaceResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetSpaceHeaders()
return self.get_space_with_options(request, headers, runtime)
async def get_space_async(
self,
request: dingtalksns_storage__1__0_models.GetSpaceRequest,
) -> dingtalksns_storage__1__0_models.GetSpaceResponse:
"""
@summary 三方个人应用获取IM会话存储空间信息
@param request: GetSpaceRequest
@return: GetSpaceResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.GetSpaceHeaders()
return await self.get_space_with_options_async(request, headers, runtime)
def list_all_dentries_with_options(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListAllDentriesRequest,
headers: dingtalksns_storage__1__0_models.ListAllDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListAllDentriesResponse:
"""
@summary 三方个人应用获取全部文件或文件夹列表
@param request: ListAllDentriesRequest
@param headers: ListAllDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListAllDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='ListAllDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/listAll',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListAllDentriesResponse(),
self.execute(params, req, runtime)
)
async def list_all_dentries_with_options_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListAllDentriesRequest,
headers: dingtalksns_storage__1__0_models.ListAllDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListAllDentriesResponse:
"""
@summary 三方个人应用获取全部文件或文件夹列表
@param request: ListAllDentriesRequest
@param headers: ListAllDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListAllDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='ListAllDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries/listAll',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListAllDentriesResponse(),
await self.execute_async(params, req, runtime)
)
def list_all_dentries(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListAllDentriesRequest,
) -> dingtalksns_storage__1__0_models.ListAllDentriesResponse:
"""
@summary 三方个人应用获取全部文件或文件夹列表
@param request: ListAllDentriesRequest
@return: ListAllDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListAllDentriesHeaders()
return self.list_all_dentries_with_options(space_id, request, headers, runtime)
async def list_all_dentries_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListAllDentriesRequest,
) -> dingtalksns_storage__1__0_models.ListAllDentriesResponse:
"""
@summary 三方个人应用获取全部文件或文件夹列表
@param request: ListAllDentriesRequest
@return: ListAllDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListAllDentriesHeaders()
return await self.list_all_dentries_with_options_async(space_id, request, headers, runtime)
def list_dentries_with_options(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListDentriesRequest,
headers: dingtalksns_storage__1__0_models.ListDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListDentriesResponse:
"""
@summary 三方个人应用获取文件列表
@param request: ListDentriesRequest
@param headers: ListDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.max_results):
query['maxResults'] = request.max_results
if not UtilClient.is_unset(request.next_token):
query['nextToken'] = request.next_token
if not UtilClient.is_unset(request.order):
query['order'] = request.order
if not UtilClient.is_unset(request.order_by):
query['orderBy'] = request.order_by
if not UtilClient.is_unset(request.parent_id):
query['parentId'] = request.parent_id
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
if not UtilClient.is_unset(request.with_thumbnail):
query['withThumbnail'] = request.with_thumbnail
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query)
)
params = open_api_models.Params(
action='ListDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries',
method='GET',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListDentriesResponse(),
self.execute(params, req, runtime)
)
async def list_dentries_with_options_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListDentriesRequest,
headers: dingtalksns_storage__1__0_models.ListDentriesHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListDentriesResponse:
"""
@summary 三方个人应用获取文件列表
@param request: ListDentriesRequest
@param headers: ListDentriesHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListDentriesResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.max_results):
query['maxResults'] = request.max_results
if not UtilClient.is_unset(request.next_token):
query['nextToken'] = request.next_token
if not UtilClient.is_unset(request.order):
query['order'] = request.order
if not UtilClient.is_unset(request.order_by):
query['orderBy'] = request.order_by
if not UtilClient.is_unset(request.parent_id):
query['parentId'] = request.parent_id
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
if not UtilClient.is_unset(request.with_thumbnail):
query['withThumbnail'] = request.with_thumbnail
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query)
)
params = open_api_models.Params(
action='ListDentries',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/spaces/{space_id}/dentries',
method='GET',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListDentriesResponse(),
await self.execute_async(params, req, runtime)
)
def list_dentries(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListDentriesRequest,
) -> dingtalksns_storage__1__0_models.ListDentriesResponse:
"""
@summary 三方个人应用获取文件列表
@param request: ListDentriesRequest
@return: ListDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListDentriesHeaders()
return self.list_dentries_with_options(space_id, request, headers, runtime)
async def list_dentries_async(
self,
space_id: str,
request: dingtalksns_storage__1__0_models.ListDentriesRequest,
) -> dingtalksns_storage__1__0_models.ListDentriesResponse:
"""
@summary 三方个人应用获取文件列表
@param request: ListDentriesRequest
@return: ListDentriesResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListDentriesHeaders()
return await self.list_dentries_with_options_async(space_id, request, headers, runtime)
def list_expired_with_options(
self,
request: dingtalksns_storage__1__0_models.ListExpiredRequest,
headers: dingtalksns_storage__1__0_models.ListExpiredHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListExpiredResponse:
"""
@summary 获取会话过期文件列表
@param request: ListExpiredRequest
@param headers: ListExpiredHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListExpiredResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.open_conversation_id):
body['openConversationId'] = request.open_conversation_id
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='ListExpired',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/conversations/expiredFileLists/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListExpiredResponse(),
self.execute(params, req, runtime)
)
async def list_expired_with_options_async(
self,
request: dingtalksns_storage__1__0_models.ListExpiredRequest,
headers: dingtalksns_storage__1__0_models.ListExpiredHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.ListExpiredResponse:
"""
@summary 获取会话过期文件列表
@param request: ListExpiredRequest
@param headers: ListExpiredHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: ListExpiredResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.open_conversation_id):
body['openConversationId'] = request.open_conversation_id
if not UtilClient.is_unset(request.option):
body['option'] = request.option
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='ListExpired',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/conversations/expiredFileLists/query',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.ListExpiredResponse(),
await self.execute_async(params, req, runtime)
)
def list_expired(
self,
request: dingtalksns_storage__1__0_models.ListExpiredRequest,
) -> dingtalksns_storage__1__0_models.ListExpiredResponse:
"""
@summary 获取会话过期文件列表
@param request: ListExpiredRequest
@return: ListExpiredResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListExpiredHeaders()
return self.list_expired_with_options(request, headers, runtime)
async def list_expired_async(
self,
request: dingtalksns_storage__1__0_models.ListExpiredRequest,
) -> dingtalksns_storage__1__0_models.ListExpiredResponse:
"""
@summary 获取会话过期文件列表
@param request: ListExpiredRequest
@return: ListExpiredResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.ListExpiredHeaders()
return await self.list_expired_with_options_async(request, headers, runtime)
def subscribe_event_with_options(
self,
request: dingtalksns_storage__1__0_models.SubscribeEventRequest,
headers: dingtalksns_storage__1__0_models.SubscribeEventHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.SubscribeEventResponse:
"""
@summary 三方个人应用订阅文件变更事件
@param request: SubscribeEventRequest
@param headers: SubscribeEventHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: SubscribeEventResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.scope):
body['scope'] = request.scope
if not UtilClient.is_unset(request.scope_id):
body['scopeId'] = request.scope_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='SubscribeEvent',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/events/subscribe',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.SubscribeEventResponse(),
self.execute(params, req, runtime)
)
async def subscribe_event_with_options_async(
self,
request: dingtalksns_storage__1__0_models.SubscribeEventRequest,
headers: dingtalksns_storage__1__0_models.SubscribeEventHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.SubscribeEventResponse:
"""
@summary 三方个人应用订阅文件变更事件
@param request: SubscribeEventRequest
@param headers: SubscribeEventHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: SubscribeEventResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.scope):
body['scope'] = request.scope
if not UtilClient.is_unset(request.scope_id):
body['scopeId'] = request.scope_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='SubscribeEvent',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/events/subscribe',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.SubscribeEventResponse(),
await self.execute_async(params, req, runtime)
)
def subscribe_event(
self,
request: dingtalksns_storage__1__0_models.SubscribeEventRequest,
) -> dingtalksns_storage__1__0_models.SubscribeEventResponse:
"""
@summary 三方个人应用订阅文件变更事件
@param request: SubscribeEventRequest
@return: SubscribeEventResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.SubscribeEventHeaders()
return self.subscribe_event_with_options(request, headers, runtime)
async def subscribe_event_async(
self,
request: dingtalksns_storage__1__0_models.SubscribeEventRequest,
) -> dingtalksns_storage__1__0_models.SubscribeEventResponse:
"""
@summary 三方个人应用订阅文件变更事件
@param request: SubscribeEventRequest
@return: SubscribeEventResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.SubscribeEventHeaders()
return await self.subscribe_event_with_options_async(request, headers, runtime)
def unsubscribe_event_with_options(
self,
request: dingtalksns_storage__1__0_models.UnsubscribeEventRequest,
headers: dingtalksns_storage__1__0_models.UnsubscribeEventHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.UnsubscribeEventResponse:
"""
@summary 三方个人应用取消订阅文件变更事件
@param request: UnsubscribeEventRequest
@param headers: UnsubscribeEventHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: UnsubscribeEventResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.scope):
body['scope'] = request.scope
if not UtilClient.is_unset(request.scope_id):
body['scopeId'] = request.scope_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='UnsubscribeEvent',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/events/unsubscribe',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.UnsubscribeEventResponse(),
self.execute(params, req, runtime)
)
async def unsubscribe_event_with_options_async(
self,
request: dingtalksns_storage__1__0_models.UnsubscribeEventRequest,
headers: dingtalksns_storage__1__0_models.UnsubscribeEventHeaders,
runtime: util_models.RuntimeOptions,
) -> dingtalksns_storage__1__0_models.UnsubscribeEventResponse:
"""
@summary 三方个人应用取消订阅文件变更事件
@param request: UnsubscribeEventRequest
@param headers: UnsubscribeEventHeaders
@param runtime: runtime options for this request RuntimeOptions
@return: UnsubscribeEventResponse
"""
UtilClient.validate_model(request)
query = {}
if not UtilClient.is_unset(request.union_id):
query['unionId'] = request.union_id
body = {}
if not UtilClient.is_unset(request.scope):
body['scope'] = request.scope
if not UtilClient.is_unset(request.scope_id):
body['scopeId'] = request.scope_id
real_headers = {}
if not UtilClient.is_unset(headers.common_headers):
real_headers = headers.common_headers
if not UtilClient.is_unset(headers.x_acs_dingtalk_access_token):
real_headers['x-acs-dingtalk-access-token'] = UtilClient.to_jsonstring(headers.x_acs_dingtalk_access_token)
req = open_api_models.OpenApiRequest(
headers=real_headers,
query=OpenApiUtilClient.query(query),
body=OpenApiUtilClient.parse_to_map(body)
)
params = open_api_models.Params(
action='UnsubscribeEvent',
version='snsStorage_1.0',
protocol='HTTP',
pathname=f'/v1.0/snsStorage/events/unsubscribe',
method='POST',
auth_type='AK',
style='ROA',
req_body_type='none',
body_type='json'
)
return TeaCore.from_map(
dingtalksns_storage__1__0_models.UnsubscribeEventResponse(),
await self.execute_async(params, req, runtime)
)
def unsubscribe_event(
self,
request: dingtalksns_storage__1__0_models.UnsubscribeEventRequest,
) -> dingtalksns_storage__1__0_models.UnsubscribeEventResponse:
"""
@summary 三方个人应用取消订阅文件变更事件
@param request: UnsubscribeEventRequest
@return: UnsubscribeEventResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.UnsubscribeEventHeaders()
return self.unsubscribe_event_with_options(request, headers, runtime)
async def unsubscribe_event_async(
self,
request: dingtalksns_storage__1__0_models.UnsubscribeEventRequest,
) -> dingtalksns_storage__1__0_models.UnsubscribeEventResponse:
"""
@summary 三方个人应用取消订阅文件变更事件
@param request: UnsubscribeEventRequest
@return: UnsubscribeEventResponse
"""
runtime = util_models.RuntimeOptions()
headers = dingtalksns_storage__1__0_models.UnsubscribeEventHeaders()
return await self.unsubscribe_event_with_options_async(request, headers, runtime)