# coding: utf-8 """ LINE Messaging API This document describes LINE Messaging API. # noqa: E501 The version of the OpenAPI document: 0.0.1 Generated by OpenAPI Generator (https://openapi-generator.tech) Do not edit the class manually. """ import re # noqa: F401 import io import warnings from pydantic.v1 import validate_arguments, ValidationError from typing_extensions import Annotated from typing import overload, Optional, Union, Awaitable from pydantic.v1 import Field, StrictStr, conint from typing import Optional from linebot.v3.module.models.acquire_chat_control_request import AcquireChatControlRequest from linebot.v3.module.models.detach_module_request import DetachModuleRequest from linebot.v3.module.models.get_modules_response import GetModulesResponse from linebot.v3.module.async_api_client import AsyncApiClient from linebot.v3.module.api_response import ApiResponse from linebot.v3.module.exceptions import ( # noqa: F401 ApiTypeError, ApiValueError ) class AsyncLineModule(object): """NOTE: This class is auto generated by OpenAPI Generator Ref: https://openapi-generator.tech Do not edit the class manually. """ def __init__(self, api_client=None): if api_client is None: api_client = AsyncApiClient.get_default() self.api_client = api_client self.line_base_path = "https://api.line.me" @overload async def acquire_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], acquire_chat_control_request : Optional[AcquireChatControlRequest] = None, **kwargs) -> None: # noqa: E501 ... @overload def acquire_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], acquire_chat_control_request : Optional[AcquireChatControlRequest] = None, async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501 ... @validate_arguments def acquire_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], acquire_chat_control_request : Optional[AcquireChatControlRequest] = None, async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501 """acquire_chat_control # noqa: E501 If the Standby Channel wants to take the initiative (Chat Control), it calls the Acquire Control API. The channel that was previously an Active Channel will automatically switch to a Standby Channel. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.acquire_chat_control(chat_id, acquire_chat_control_request, async_req=True) >>> result = thread.get() :param chat_id: The `userId`, `roomId`, or `groupId` (required) :type chat_id: str :param acquire_chat_control_request: :type acquire_chat_control_request: AcquireChatControlRequest :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ kwargs['_return_http_data_only'] = True if '_preload_content' in kwargs: raise ValueError("Error! Please call the acquire_chat_control_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data") if async_req is not None: kwargs['async_req'] = async_req return self.acquire_chat_control_with_http_info(chat_id, acquire_chat_control_request, **kwargs) # noqa: E501 @validate_arguments def acquire_chat_control_with_http_info(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], acquire_chat_control_request : Optional[AcquireChatControlRequest] = None, **kwargs) -> ApiResponse: # noqa: E501 """acquire_chat_control # noqa: E501 If the Standby Channel wants to take the initiative (Chat Control), it calls the Acquire Control API. The channel that was previously an Active Channel will automatically switch to a Standby Channel. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.acquire_chat_control_with_http_info(chat_id, acquire_chat_control_request, async_req=True) >>> result = thread.get() :param chat_id: The `userId`, `roomId`, or `groupId` (required) :type chat_id: str :param acquire_chat_control_request: :type acquire_chat_control_request: AcquireChatControlRequest :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _preload_content: if False, the ApiResponse.data will be set to none and raw_data will store the HTTP response body without reading/decoding. Default is True. :type _preload_content: bool, optional :param _return_http_data_only: response data instead of ApiResponse object with status code, headers, etc :type _return_http_data_only: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :type _content_type: string, optional: force content-type for the request :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ _host = self.line_base_path _params = locals() _all_params = [ 'chat_id', 'acquire_chat_control_request' ] _all_params.extend( [ 'async_req', '_return_http_data_only', '_preload_content', '_request_timeout', '_request_auth', '_content_type', '_headers' ] ) # validate the arguments for _key, _val in _params['kwargs'].items(): if _key not in _all_params: raise ApiTypeError( "Got an unexpected keyword argument '%s'" " to method acquire_chat_control" % _key ) _params[_key] = _val del _params['kwargs'] _collection_formats = {} # process the path parameters _path_params = {} if _params['chat_id']: _path_params['chatId'] = _params['chat_id'] # process the query parameters _query_params = [] # process the header parameters _header_params = dict(_params.get('_headers', {})) # process the form parameters _form_params = [] _files = {} # process the body parameter _body_params = None if _params['acquire_chat_control_request'] is not None: _body_params = _params['acquire_chat_control_request'] # set the HTTP header `Content-Type` _content_types_list = _params.get('_content_type', self.api_client.select_header_content_type( ['application/json'])) if _content_types_list: _header_params['Content-Type'] = _content_types_list # authentication setting _auth_settings = ['Bearer'] # noqa: E501 _response_types_map = {} return self.api_client.call_api( '/v2/bot/chat/{chatId}/control/acquire', 'POST', _path_params, _query_params, _header_params, body=_body_params, post_params=_form_params, files=_files, response_types_map=_response_types_map, auth_settings=_auth_settings, async_req=_params.get('async_req'), _return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501 _preload_content=_params.get('_preload_content', True), _request_timeout=_params.get('_request_timeout'), _host=_host, collection_formats=_collection_formats, _request_auth=_params.get('_request_auth')) @overload async def detach_module(self, detach_module_request : Optional[DetachModuleRequest] = None, **kwargs) -> None: # noqa: E501 ... @overload def detach_module(self, detach_module_request : Optional[DetachModuleRequest] = None, async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501 ... @validate_arguments def detach_module(self, detach_module_request : Optional[DetachModuleRequest] = None, async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501 """detach_module # noqa: E501 The module channel admin calls the Detach API to detach the module channel from a LINE Official Account. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.detach_module(detach_module_request, async_req=True) >>> result = thread.get() :param detach_module_request: :type detach_module_request: DetachModuleRequest :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ kwargs['_return_http_data_only'] = True if '_preload_content' in kwargs: raise ValueError("Error! Please call the detach_module_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data") if async_req is not None: kwargs['async_req'] = async_req return self.detach_module_with_http_info(detach_module_request, **kwargs) # noqa: E501 @validate_arguments def detach_module_with_http_info(self, detach_module_request : Optional[DetachModuleRequest] = None, **kwargs) -> ApiResponse: # noqa: E501 """detach_module # noqa: E501 The module channel admin calls the Detach API to detach the module channel from a LINE Official Account. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.detach_module_with_http_info(detach_module_request, async_req=True) >>> result = thread.get() :param detach_module_request: :type detach_module_request: DetachModuleRequest :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _preload_content: if False, the ApiResponse.data will be set to none and raw_data will store the HTTP response body without reading/decoding. Default is True. :type _preload_content: bool, optional :param _return_http_data_only: response data instead of ApiResponse object with status code, headers, etc :type _return_http_data_only: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :type _content_type: string, optional: force content-type for the request :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ _host = self.line_base_path _params = locals() _all_params = [ 'detach_module_request' ] _all_params.extend( [ 'async_req', '_return_http_data_only', '_preload_content', '_request_timeout', '_request_auth', '_content_type', '_headers' ] ) # validate the arguments for _key, _val in _params['kwargs'].items(): if _key not in _all_params: raise ApiTypeError( "Got an unexpected keyword argument '%s'" " to method detach_module" % _key ) _params[_key] = _val del _params['kwargs'] _collection_formats = {} # process the path parameters _path_params = {} # process the query parameters _query_params = [] # process the header parameters _header_params = dict(_params.get('_headers', {})) # process the form parameters _form_params = [] _files = {} # process the body parameter _body_params = None if _params['detach_module_request'] is not None: _body_params = _params['detach_module_request'] # set the HTTP header `Content-Type` _content_types_list = _params.get('_content_type', self.api_client.select_header_content_type( ['application/json'])) if _content_types_list: _header_params['Content-Type'] = _content_types_list # authentication setting _auth_settings = ['Bearer'] # noqa: E501 _response_types_map = {} return self.api_client.call_api( '/v2/bot/channel/detach', 'POST', _path_params, _query_params, _header_params, body=_body_params, post_params=_form_params, files=_files, response_types_map=_response_types_map, auth_settings=_auth_settings, async_req=_params.get('async_req'), _return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501 _preload_content=_params.get('_preload_content', True), _request_timeout=_params.get('_request_timeout'), _host=_host, collection_formats=_collection_formats, _request_auth=_params.get('_request_auth')) @overload async def get_modules(self, start : Annotated[Optional[StrictStr], Field(description="Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. ")] = None, limit : Annotated[Optional[conint(strict=True, le=100)], Field(description="Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 ")] = None, **kwargs) -> GetModulesResponse: # noqa: E501 ... @overload def get_modules(self, start : Annotated[Optional[StrictStr], Field(description="Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. ")] = None, limit : Annotated[Optional[conint(strict=True, le=100)], Field(description="Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 ")] = None, async_req: Optional[bool]=True, **kwargs) -> GetModulesResponse: # noqa: E501 ... @validate_arguments def get_modules(self, start : Annotated[Optional[StrictStr], Field(description="Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. ")] = None, limit : Annotated[Optional[conint(strict=True, le=100)], Field(description="Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 ")] = None, async_req: Optional[bool]=None, **kwargs) -> Union[GetModulesResponse, Awaitable[GetModulesResponse]]: # noqa: E501 """get_modules # noqa: E501 Gets a list of basic information about the bots of multiple LINE Official Accounts that have attached module channels. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.get_modules(start, limit, async_req=True) >>> result = thread.get() :param start: Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. :type start: str :param limit: Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 :type limit: int :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: GetModulesResponse """ kwargs['_return_http_data_only'] = True if '_preload_content' in kwargs: raise ValueError("Error! Please call the get_modules_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data") if async_req is not None: kwargs['async_req'] = async_req return self.get_modules_with_http_info(start, limit, **kwargs) # noqa: E501 @validate_arguments def get_modules_with_http_info(self, start : Annotated[Optional[StrictStr], Field(description="Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. ")] = None, limit : Annotated[Optional[conint(strict=True, le=100)], Field(description="Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 ")] = None, **kwargs) -> ApiResponse: # noqa: E501 """get_modules # noqa: E501 Gets a list of basic information about the bots of multiple LINE Official Accounts that have attached module channels. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.get_modules_with_http_info(start, limit, async_req=True) >>> result = thread.get() :param start: Value of the continuation token found in the next property of the JSON object returned in the response. If you can't get all basic information about the bots in one request, include this parameter to get the remaining array. :type start: str :param limit: Specify the maximum number of bots that you get basic information from. The default value is 100. Max value: 100 :type limit: int :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _preload_content: if False, the ApiResponse.data will be set to none and raw_data will store the HTTP response body without reading/decoding. Default is True. :type _preload_content: bool, optional :param _return_http_data_only: response data instead of ApiResponse object with status code, headers, etc :type _return_http_data_only: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :type _content_type: string, optional: force content-type for the request :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: tuple(GetModulesResponse, status_code(int), headers(HTTPHeaderDict)) """ _host = self.line_base_path _params = locals() _all_params = [ 'start', 'limit' ] _all_params.extend( [ 'async_req', '_return_http_data_only', '_preload_content', '_request_timeout', '_request_auth', '_content_type', '_headers' ] ) # validate the arguments for _key, _val in _params['kwargs'].items(): if _key not in _all_params: raise ApiTypeError( "Got an unexpected keyword argument '%s'" " to method get_modules" % _key ) _params[_key] = _val del _params['kwargs'] _collection_formats = {} # process the path parameters _path_params = {} # process the query parameters _query_params = [] if _params.get('start') is not None: # noqa: E501 _query_params.append(('start', _params['start'])) if _params.get('limit') is not None: # noqa: E501 _query_params.append(('limit', _params['limit'])) # process the header parameters _header_params = dict(_params.get('_headers', {})) # process the form parameters _form_params = [] _files = {} # process the body parameter _body_params = None # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( ['application/json']) # noqa: E501 # authentication setting _auth_settings = ['Bearer'] # noqa: E501 _response_types_map = { '200': "GetModulesResponse", } return self.api_client.call_api( '/v2/bot/list', 'GET', _path_params, _query_params, _header_params, body=_body_params, post_params=_form_params, files=_files, response_types_map=_response_types_map, auth_settings=_auth_settings, async_req=_params.get('async_req'), _return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501 _preload_content=_params.get('_preload_content', True), _request_timeout=_params.get('_request_timeout'), _host=_host, collection_formats=_collection_formats, _request_auth=_params.get('_request_auth')) @overload async def release_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], **kwargs) -> None: # noqa: E501 ... @overload def release_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], async_req: Optional[bool]=True, **kwargs) -> None: # noqa: E501 ... @validate_arguments def release_chat_control(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], async_req: Optional[bool]=None, **kwargs) -> Union[None, Awaitable[None]]: # noqa: E501 """release_chat_control # noqa: E501 To return the initiative (Chat Control) of Active Channel to Primary Channel, call the Release Control API. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.release_chat_control(chat_id, async_req=True) >>> result = thread.get() :param chat_id: The `userId`, `roomId`, or `groupId` (required) :type chat_id: str :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ kwargs['_return_http_data_only'] = True if '_preload_content' in kwargs: raise ValueError("Error! Please call the release_chat_control_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data") if async_req is not None: kwargs['async_req'] = async_req return self.release_chat_control_with_http_info(chat_id, **kwargs) # noqa: E501 @validate_arguments def release_chat_control_with_http_info(self, chat_id : Annotated[StrictStr, Field(..., description="The `userId`, `roomId`, or `groupId`")], **kwargs) -> ApiResponse: # noqa: E501 """release_chat_control # noqa: E501 To return the initiative (Chat Control) of Active Channel to Primary Channel, call the Release Control API. # noqa: E501 This method makes a synchronous HTTP request by default. To make an asynchronous HTTP request, please pass async_req=True >>> thread = api.release_chat_control_with_http_info(chat_id, async_req=True) >>> result = thread.get() :param chat_id: The `userId`, `roomId`, or `groupId` (required) :type chat_id: str :param async_req: Whether to execute the request asynchronously. :type async_req: bool, optional :param _preload_content: if False, the ApiResponse.data will be set to none and raw_data will store the HTTP response body without reading/decoding. Default is True. :type _preload_content: bool, optional :param _return_http_data_only: response data instead of ApiResponse object with status code, headers, etc :type _return_http_data_only: bool, optional :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :param _request_auth: set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request. :type _request_auth: dict, optional :type _content_type: string, optional: force content-type for the request :return: Returns the result object. If the method is called asynchronously, returns the request thread. :rtype: None """ _host = self.line_base_path _params = locals() _all_params = [ 'chat_id' ] _all_params.extend( [ 'async_req', '_return_http_data_only', '_preload_content', '_request_timeout', '_request_auth', '_content_type', '_headers' ] ) # validate the arguments for _key, _val in _params['kwargs'].items(): if _key not in _all_params: raise ApiTypeError( "Got an unexpected keyword argument '%s'" " to method release_chat_control" % _key ) _params[_key] = _val del _params['kwargs'] _collection_formats = {} # process the path parameters _path_params = {} if _params['chat_id']: _path_params['chatId'] = _params['chat_id'] # process the query parameters _query_params = [] # process the header parameters _header_params = dict(_params.get('_headers', {})) # process the form parameters _form_params = [] _files = {} # process the body parameter _body_params = None # authentication setting _auth_settings = ['Bearer'] # noqa: E501 _response_types_map = {} return self.api_client.call_api( '/v2/bot/chat/{chatId}/control/release', 'POST', _path_params, _query_params, _header_params, body=_body_params, post_params=_form_params, files=_files, response_types_map=_response_types_map, auth_settings=_auth_settings, async_req=_params.get('async_req'), _return_http_data_only=_params.get('_return_http_data_only'), # noqa: E501 _preload_content=_params.get('_preload_content', True), _request_timeout=_params.get('_request_timeout'), _host=_host, collection_formats=_collection_formats, _request_auth=_params.get('_request_auth'))