# coding: utf-8 """ ThousandEyes for OpenTelemetry API ThousandEyes for OpenTelemetry provides machine-to-machine integration between ThousandEyes and its customers. It allows you to export ThousandEyes telemetry data in OTel format, which is widely used in the industry. With ThousandEyes for OTel, you can leverage frameworks widely used in the observability domain - such as Splunk, Grafana, and Honeycomb - to capture and analyze ThousandEyes data. Any client that supports OTel can use ThousandEyes for OpenTelemetry. ThousandEyes for OTel is made up of the following components: * Data streaming APIs that you can use to configure and enable your ThousandEyes tests with OTel-compatible streams, in particular to configure how ThousandEyes telemetry data is exported to client integrations. * A set of streaming pipelines called _collectors_ that actively fetch ThousandEyes network test data, enrich the data with some additional detail, filter, and push the data to the customer-configured endpoints, depending on what you configure via the public APIs. * Third-party OTel collectors that receive, transform, filter, and export different metrics to client applications such as AppD, or any other OTel-capable client configuration. For more information about ThousandEyes for OpenTelemetry, see the [documentation](https://docs.thousandeyes.com/product-documentation/api/opentelemetry). Generated by OpenAPI Generator (https://openapi-generator.tech) Do not edit the class manually. """ # noqa: E501 import warnings from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt from typing import Any, Dict, List, Optional, Tuple, Union from typing_extensions import Annotated from importlib.metadata import version import thousandeyes_sdk.streaming.models from pydantic import Field, StrictStr from typing import List, Optional from typing_extensions import Annotated from thousandeyes_sdk.streaming.models.create_stream_response import CreateStreamResponse from thousandeyes_sdk.streaming.models.get_stream_response import GetStreamResponse from thousandeyes_sdk.streaming.models.put_stream import PutStream from thousandeyes_sdk.streaming.models.stream import Stream from thousandeyes_sdk.streaming.models.stream_type import StreamType from thousandeyes_sdk.core.api_client import ApiClient, RequestSerialized from thousandeyes_sdk.core.api_response import ApiResponse from thousandeyes_sdk.core.rest import RESTResponseType class StreamingApi: """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) -> None: if api_client is None: api_client = ApiClient.get_default() api_client.user_agent = "ThousandEyesSDK-Python/{0}".format(version("thousandeyes-sdk-streaming")) self.api_client = api_client @validate_call def create_stream( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, stream: Annotated[Optional[Stream], Field(description="Stream to configure")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> CreateStreamResponse: """Create data stream Creates a new data stream. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param stream: Stream to configure :type stream: Stream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._create_stream_serialize( aid=aid, stream=stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '201': "CreateStreamResponse", '400': "BadRequestError", '401': "UnauthorizedError", '409': None, '412': "ApiErrorIntegrationLimits", '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ).data @validate_call def create_stream_with_http_info( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, stream: Annotated[Optional[Stream], Field(description="Stream to configure")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[CreateStreamResponse]: """Create data stream Creates a new data stream. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param stream: Stream to configure :type stream: Stream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._create_stream_serialize( aid=aid, stream=stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '201': "CreateStreamResponse", '400': "BadRequestError", '401': "UnauthorizedError", '409': None, '412': "ApiErrorIntegrationLimits", '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ) @validate_call def create_stream_without_preload_content( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, stream: Annotated[Optional[Stream], Field(description="Stream to configure")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Create data stream Creates a new data stream. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param stream: Stream to configure :type stream: Stream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._create_stream_serialize( aid=aid, stream=stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '201': "CreateStreamResponse", '400': "BadRequestError", '401': "UnauthorizedError", '409': None, '412': "ApiErrorIntegrationLimits", '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _create_stream_serialize( self, aid, stream, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[str, Union[str, bytes]] = {} _body_params: Optional[bytes] = None # process the path parameters # process the query parameters if aid is not None: _query_params.append(('aid', aid)) # process the header parameters # process the form parameters # process the body parameter if stream is not None: _body_params = stream # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/hal+json', 'application/json', 'application/problem+json' ] ) # set the HTTP header `Content-Type` if _content_type: _header_params['Content-Type'] = _content_type else: _default_content_type = ( self.api_client.select_header_content_type( [ 'application/json' ] ) ) if _default_content_type is not None: _header_params['Content-Type'] = _default_content_type # authentication setting _auth_settings: List[str] = [ 'BearerAuth' ] return self.api_client.param_serialize( method='POST', resource_path='/stream', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call def delete_stream( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> None: """Delete a data stream Deletes a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._delete_stream_serialize( id=id, aid=aid, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '204': None, '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ).data @validate_call def delete_stream_with_http_info( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[None]: """Delete a data stream Deletes a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._delete_stream_serialize( id=id, aid=aid, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '204': None, '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ) @validate_call def delete_stream_without_preload_content( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Delete a data stream Deletes a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._delete_stream_serialize( id=id, aid=aid, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '204': None, '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _delete_stream_serialize( self, id, aid, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[str, Union[str, bytes]] = {} _body_params: Optional[bytes] = None # process the path parameters if id is not None: _path_params['id'] = id # process the query parameters if aid is not None: _query_params.append(('aid', aid)) # process the header parameters # process the form parameters # process the body parameter # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/json', 'application/problem+json' ] ) # authentication setting _auth_settings: List[str] = [ 'BearerAuth' ] return self.api_client.param_serialize( method='DELETE', resource_path='/stream/{id}', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call def get_stream( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> GetStreamResponse: """Retrieve data stream Retrieves a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_stream_serialize( id=id, aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ).data @validate_call def get_stream_with_http_info( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[GetStreamResponse]: """Retrieve data stream Retrieves a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_stream_serialize( id=id, aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ) @validate_call def get_stream_without_preload_content( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Retrieve data stream Retrieves a configured data stream using its ID. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_stream_serialize( id=id, aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _get_stream_serialize( self, id, aid, type, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[str, Union[str, bytes]] = {} _body_params: Optional[bytes] = None # process the path parameters if id is not None: _path_params['id'] = id # process the query parameters if aid is not None: _query_params.append(('aid', aid)) if type is not None: _query_params.append(('type', type.value)) # process the header parameters # process the form parameters # process the body parameter # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/hal+json', 'application/json', 'application/problem+json' ] ) # authentication setting _auth_settings: List[str] = [ 'BearerAuth' ] return self.api_client.param_serialize( method='GET', resource_path='/stream/{id}', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call def get_streams( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> List[GetStreamResponse]: """List data streams Retrieves a list of configured data streams. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_streams_serialize( aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "List[GetStreamResponse]", '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ).data @validate_call def get_streams_with_http_info( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[List[GetStreamResponse]]: """List data streams Retrieves a list of configured data streams. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_streams_serialize( aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "List[GetStreamResponse]", '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ) @validate_call def get_streams_without_preload_content( self, aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, type: Annotated[Optional[StreamType], Field(description="Optional filter on type of Stream; should match one of Stream `type` enum")] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """List data streams Retrieves a list of configured data streams. :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param type: Optional filter on type of Stream; should match one of Stream `type` enum :type type: StreamType :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._get_streams_serialize( aid=aid, type=type, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "List[GetStreamResponse]", '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _get_streams_serialize( self, aid, type, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[str, Union[str, bytes]] = {} _body_params: Optional[bytes] = None # process the path parameters # process the query parameters if aid is not None: _query_params.append(('aid', aid)) if type is not None: _query_params.append(('type', type.value)) # process the header parameters # process the form parameters # process the body parameter # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/hal+json', 'application/json', 'application/problem+json' ] ) # authentication setting _auth_settings: List[str] = [ 'BearerAuth' ] return self.api_client.param_serialize( method='GET', resource_path='/stream', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth ) @validate_call def update_stream( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, put_stream: Optional[PutStream] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> GetStreamResponse: """Update data stream Updates a configured data stream using its ID. The fields are overwritten, not appended. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param put_stream: :type put_stream: PutStream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._update_stream_serialize( id=id, aid=aid, put_stream=put_stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '204': None, '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ).data @validate_call def update_stream_with_http_info( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, put_stream: Optional[PutStream] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> ApiResponse[GetStreamResponse]: """Update data stream Updates a configured data stream using its ID. The fields are overwritten, not appended. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param put_stream: :type put_stream: PutStream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._update_stream_serialize( id=id, aid=aid, put_stream=put_stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '204': None, '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) response_data.read() return self.api_client.response_deserialize( response_data=response_data, response_types_map=_response_types_map, models=thousandeyes_sdk.streaming.models, ) @validate_call def update_stream_without_preload_content( self, id: Annotated[StrictStr, Field(description="ID of stream to query")], aid: Annotated[Optional[StrictStr], Field(description="A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response.")] = None, put_stream: Optional[PutStream] = None, _request_timeout: Union[ None, Annotated[StrictFloat, Field(gt=0)], Tuple[ Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)] ] ] = None, _request_auth: Optional[Dict[StrictStr, Any]] = None, _content_type: Optional[StrictStr] = None, _headers: Optional[Dict[StrictStr, Any]] = None, _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, ) -> RESTResponseType: """Update data stream Updates a configured data stream using its ID. The fields are overwritten, not appended. :param id: ID of stream to query (required) :type id: str :param aid: A unique identifier associated with your account group. You can retrieve your `AccountGroupId` from the `/account-groups` endpoint. Note that you must be assigned to the target account group. Specifying this parameter without being assigned to the target account group will result in an error response. :type aid: str :param put_stream: :type put_stream: PutStream :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. :type _request_timeout: int, tuple(int, int), optional :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 :param _content_type: force content-type for the request. :type _content_type: str, Optional :param _headers: set to override the headers for a single request; this effectively ignores the headers in the spec for a single request. :type _headers: dict, optional :param _host_index: set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request. :type _host_index: int, optional :return: Returns the result object. """ # noqa: E501 _param = self._update_stream_serialize( id=id, aid=aid, put_stream=put_stream, _request_auth=_request_auth, _content_type=_content_type, _headers=_headers, _host_index=_host_index ) _response_types_map: Dict[str, Optional[str]] = { '200': "GetStreamResponse", '204': None, '400': "BadRequestError", '401': "UnauthorizedError", '403': None, '404': None, '429': None, '500': "ApiError", } response_data = self.api_client.call_api( *_param, _request_timeout=_request_timeout ) return response_data.response def _update_stream_serialize( self, id, aid, put_stream, _request_auth, _content_type, _headers, _host_index, ) -> RequestSerialized: _host = None _collection_formats: Dict[str, str] = { } _path_params: Dict[str, str] = {} _query_params: List[Tuple[str, str]] = [] _header_params: Dict[str, Optional[str]] = _headers or {} _form_params: List[Tuple[str, str]] = [] _files: Dict[str, Union[str, bytes]] = {} _body_params: Optional[bytes] = None # process the path parameters if id is not None: _path_params['id'] = id # process the query parameters if aid is not None: _query_params.append(('aid', aid)) # process the header parameters # process the form parameters # process the body parameter if put_stream is not None: _body_params = put_stream # set the HTTP header `Accept` _header_params['Accept'] = self.api_client.select_header_accept( [ 'application/hal+json', 'application/json', 'application/problem+json' ] ) # set the HTTP header `Content-Type` if _content_type: _header_params['Content-Type'] = _content_type else: _default_content_type = ( self.api_client.select_header_content_type( [ 'application/json' ] ) ) if _default_content_type is not None: _header_params['Content-Type'] = _default_content_type # authentication setting _auth_settings: List[str] = [ 'BearerAuth' ] return self.api_client.param_serialize( method='PUT', resource_path='/stream/{id}', path_params=_path_params, query_params=_query_params, header_params=_header_params, body=_body_params, post_params=_form_params, files=_files, auth_settings=_auth_settings, collection_formats=_collection_formats, _host=_host, _request_auth=_request_auth )