list_integrations async ¶

async def list_integrations(
    self,
    *,
    config_overrides: Optional[AsyncCloudWatchLogsClientConfig] = None,
    integration_name_prefix: Optional[
        "capo_cloudwatch_logs.types.integration_name_prefix.IntegrationNamePrefix"
    ] = None,
    integration_type: Optional[
        "capo_cloudwatch_logs.types.integration_type.IntegrationType"
    ] = None,
    integration_status: Optional[
        "capo_cloudwatch_logs.types.integration_status.IntegrationStatus"
    ] = None,
) -> (
    "capo_cloudwatch_logs.types.list_integrations_response.ListIntegrationsResponse"
):
    """<p>Returns a list of integrations between CloudWatch Logs and other services in this account. Currently, only one integration can be created in an account, and this integration must be with OpenSearch Service.</p>

    Args:
        integration_name_prefix: <p>To limit the results to integrations that start with a certain name prefix, specify that name prefix here.</p>
        integration_type: <p>To limit the results to integrations of a certain type, specify that type here.</p>
        integration_status: <p>To limit the results to integrations with a certain status, specify that status here.</p>

    Raises:
        capo_cloudwatch_logs.errors.invalid_parameter_exception.InvalidParameterException: <p>A parameter is specified incorrectly.</p>
        capo_cloudwatch_logs.errors.service_unavailable_exception.ServiceUnavailableException: <p>The service cannot complete the request.</p>
        capo_cloudwatch_logs.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_cloudwatch_logs.types.list_integrations_request.ListIntegrationsRequest]",
    ) -> AsyncOperationResponse[
        "capo_cloudwatch_logs.types.list_integrations_response.ListIntegrationsResponse"
    ]:
        import capo_cloudwatch_logs._operations.logs_20140328.list_integrations

        (
            output,
            http_response,
        ) = await capo_cloudwatch_logs._operations.logs_20140328.list_integrations.async_list_integrations(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("ListIntegrations")
    )
    input_: capo_cloudwatch_logs.types.list_integrations_request.ListIntegrationsRequest = {}
    if integration_name_prefix is not None:
        input_["integration_name_prefix"] = integration_name_prefix
    if integration_type is not None:
        input_["integration_type"] = integration_type
    if integration_status is not None:
        input_["integration_status"] = integration_status

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output