diff --git a/onelens_backend_client_v2/models.py b/onelens_backend_client_v2/models.py index 6d8cc632..ac46c8c5 100644 --- a/onelens_backend_client_v2/models.py +++ b/onelens_backend_client_v2/models.py @@ -1,6 +1,6 @@ # generated by datamodel-codegen: # filename: -# timestamp: 2024-10-24T14:06:49+00:00 +# timestamp: 2024-10-25T05:55:44+00:00 from __future__ import annotations @@ -481,6 +481,40 @@ class ConnectionType(str, Enum): REPLICA = "REPLICA" +class CostAnalyzerGraphRequest(BaseModel): + filters: List[Dict[str, Any]] = Field( + ..., description="Filters for the cost analyzer graph", title="Filters" + ) + group_by: List[str] = Field( + ..., description="Group by for the cost analyzer graph", title="Group By" + ) + + +class CostAnalyzerGraphResponse(BaseModel): + data: List[Dict[str, Any]] = Field( + ..., + description="List of dictionaries containing cost analyzer stats", + title="Data", + ) + + +class CostAnalyzerStatsRequest(BaseModel): + filters: List[Dict[str, Any]] = Field( + ..., description="Filters for the cost analyzer stats", title="Filters" + ) + group_by: List[str] = Field( + ..., description="Group by for the cost analyzer stats", title="Group By" + ) + + +class CostAnalyzerStatsResponse(BaseModel): + data: List[Dict[str, Any]] = Field( + ..., + description="List of dictionaries containing cost analyzer stats", + title="Data", + ) + + class CreateActionTypePullRequest(BaseModel): title: str = Field(..., description="The title of the pull request.", title="Title") description: str = Field( @@ -2935,6 +2969,8 @@ class OnelensDomainUtilitiesRepositoriesDynamicFiltersOperator(str, Enum): is_empty = "is_empty" is_not_empty = "is_not_empty" is_daterange_template = "is_daterange_template" + is_null = "is_null" + is_not_null = "is_not_null" class OnelensModelsServiceInterfacesPoliciesCommonsOperator(str, Enum): @@ -2948,6 +2984,8 @@ class OnelensModelsServiceInterfacesPoliciesCommonsOperator(str, Enum): less_than_or_equals = "less_than_or_equals" is_null = "is_null" is_not_null = "is_not_null" + is_empty = "is_empty" + is_not_empty = "is_not_empty" class OnelensModelsServiceInterfacesTenantDataCurServiceGranularityUnit(str, Enum): @@ -2979,6 +3017,8 @@ class OnelensModelsServiceInterfacesTenantDataMetricsServiceOperator(str, Enum): is_not = "is_not" is_null = "is_null" is_not_null = "is_not_null" + is_empty = "is_empty" + is_not_empty = "is_not_empty" class OnelensModelsServiceInterfacesTenantMetadataCommonsHierarchyNodeCategory1( @@ -3296,6 +3336,30 @@ class CloudMetadataFilters(BaseModel): ) +class CostAnalyzerTableRequest(BaseModel): + pagination: Optional[PaginationParams] = Field( + default_factory=lambda: PaginationParams.model_validate( + {"page": 1, "page_size": 10} + ), + description="Pagination parameters for the request.", + ) + filters: List[Dict[str, Any]] = Field( + ..., description="Filters for the cost analyzer table", title="Filters" + ) + group_by: List[str] = Field( + ..., description="Group by for the cost analyzer table", title="Group By" + ) + + +class CostAnalyzerTableResponse(BaseModel): + pagination: PaginationFields = Field(..., description="Pagination fields.") + data: List[Dict[str, Any]] = Field( + ..., + description="List of dictionaries containing cost analyzer stats", + title="Data", + ) + + class CreateActionTypeRequest(BaseModel): service: Union[AwsService, GcpService] = Field( ..., description="Service AWS etc.", title="Service" diff --git a/onelens_backend_client_v2/rpc/__init__.py b/onelens_backend_client_v2/rpc/__init__.py index 1a08f903..68f0f3d7 100644 --- a/onelens_backend_client_v2/rpc/__init__.py +++ b/onelens_backend_client_v2/rpc/__init__.py @@ -86,6 +86,10 @@ RecommendationUnitMigrationServiceRpcHandler, ) +from onelens_backend_client_v2.rpc.cost_analyzer_service_rpc_handler import ( + CostAnalyzerServiceRpcHandler, +) + from onelens_backend_client_v2.rpc.cloud_account_metadata_service_rpc_handler import ( CloudAccountMetadataServiceRpcHandler, ) @@ -126,6 +130,7 @@ "SavedViewServiceRpcHandler", "RecommendationUnitServiceRpcHandler", "RecommendationUnitMigrationServiceRpcHandler", + "CostAnalyzerServiceRpcHandler", "CloudAccountMetadataServiceRpcHandler", "NaviraServiceRpcHandler", "ActionTypeMigrationServiceRpcHandler", diff --git a/onelens_backend_client_v2/rpc/cost_analyzer_service_rpc_handler.py b/onelens_backend_client_v2/rpc/cost_analyzer_service_rpc_handler.py new file mode 100644 index 00000000..a774bf10 --- /dev/null +++ b/onelens_backend_client_v2/rpc/cost_analyzer_service_rpc_handler.py @@ -0,0 +1,398 @@ +# CostAnalyzerServiceRpcHandler API + +from pydantic import validate_call, Field, StrictFloat, StrictStr, StrictInt +from typing import Any, Dict, List, Optional, Tuple, Union +from typing_extensions import Annotated + + +from onelens_backend_client_v2.models import CostAnalyzerGraphRequest + + +from onelens_backend_client_v2.models import CostAnalyzerGraphResponse + + +from onelens_backend_client_v2.models import CostAnalyzerStatsRequest + + +from onelens_backend_client_v2.models import CostAnalyzerStatsResponse + + +from onelens_backend_client_v2.models import CostAnalyzerTableRequest + + +from onelens_backend_client_v2.models import CostAnalyzerTableResponse + + +from onelens_backend_client_v2.api_client import ApiClient, RequestSerialized + + +class CostAnalyzerServiceRpcHandler: + """NOTE: This class is auto generated. Do not edit the class manually.""" + + def __init__(self, api_client=None) -> None: + if api_client is None: + api_client = ApiClient.get_default() + self.api_client = api_client + + @validate_call + def fetch_graph_for_cost_analyzer( + self, + request: CostAnalyzerGraphRequest, + _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, + ) -> CostAnalyzerGraphResponse: + """Fetch Graph For Cost Analyzer + + + + :param request: (required) + :type request: CostAnalyzerGraphRequest + :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. + """ + + _param = self._fetch_graph_for_cost_analyzer_serialize( + request=request, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index, + ) + + _response_types_map: Dict[str, Optional[str]] = { + "200": "CostAnalyzerGraphResponse", + "422": "HTTPValidationError", + } + 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, + ).data + + def _fetch_graph_for_cost_analyzer_serialize( + self, + request: CostAnalyzerGraphRequest, + _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 body parameter + if request is not None: + _body_params = request + + # set the HTTP header `Accept` + _header_params["Accept"] = self.api_client.select_header_accept( + ["application/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] = [] + + return self.api_client.param_serialize( + method="POST", + resource_path="/rpc/cost_analyzer_service/fetch_graph_for_cost_analyzer", + 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 fetch_stats_for_cost_analyzer( + self, + request: CostAnalyzerStatsRequest, + _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, + ) -> CostAnalyzerStatsResponse: + """Fetch Stats For Cost Analyzer + + + + :param request: (required) + :type request: CostAnalyzerStatsRequest + :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. + """ + + _param = self._fetch_stats_for_cost_analyzer_serialize( + request=request, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index, + ) + + _response_types_map: Dict[str, Optional[str]] = { + "200": "CostAnalyzerStatsResponse", + "422": "HTTPValidationError", + } + 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, + ).data + + def _fetch_stats_for_cost_analyzer_serialize( + self, + request: CostAnalyzerStatsRequest, + _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 body parameter + if request is not None: + _body_params = request + + # set the HTTP header `Accept` + _header_params["Accept"] = self.api_client.select_header_accept( + ["application/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] = [] + + return self.api_client.param_serialize( + method="POST", + resource_path="/rpc/cost_analyzer_service/fetch_stats_for_cost_analyzer", + 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 fetch_table_for_cost_analyzer( + self, + request: CostAnalyzerTableRequest, + _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, + ) -> CostAnalyzerTableResponse: + """Fetch Table For Cost Analyzer + + + + :param request: (required) + :type request: CostAnalyzerTableRequest + :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. + """ + + _param = self._fetch_table_for_cost_analyzer_serialize( + request=request, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index, + ) + + _response_types_map: Dict[str, Optional[str]] = { + "200": "CostAnalyzerTableResponse", + "422": "HTTPValidationError", + } + 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, + ).data + + def _fetch_table_for_cost_analyzer_serialize( + self, + request: CostAnalyzerTableRequest, + _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 body parameter + if request is not None: + _body_params = request + + # set the HTTP header `Accept` + _header_params["Accept"] = self.api_client.select_header_accept( + ["application/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] = [] + + return self.api_client.param_serialize( + method="POST", + resource_path="/rpc/cost_analyzer_service/fetch_table_for_cost_analyzer", + 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, + )