From 60ebc6de9c7a112c8d382911adaefeeea646e859 Mon Sep 17 00:00:00 2001 From: pranjal Date: Thu, 21 Nov 2024 15:07:41 +0530 Subject: [PATCH] added models and rpc calls for notification service --- onelens_backend_client_v2/models.py | 22 ++- onelens_backend_client_v2/rpc/__init__.py | 5 + .../rpc/notification_service_rpc_handler.py | 144 ++++++++++++++++++ 3 files changed, 170 insertions(+), 1 deletion(-) create mode 100644 onelens_backend_client_v2/rpc/notification_service_rpc_handler.py diff --git a/onelens_backend_client_v2/models.py b/onelens_backend_client_v2/models.py index b1bd0e05..491f6d5b 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-11-20T17:29:35+00:00 +# timestamp: 2024-11-21T09:34:38+00:00 from __future__ import annotations @@ -1640,6 +1640,15 @@ class NaviraLogUpdateRequest(BaseModel): ) +class NotificationEntityType(str, Enum): + ticket = "ticket" + + +class NotificationType(str, Enum): + ticket_updated = "ticket_updated" + ticket_processed = "ticket_processed" + + class OrItem(BaseModel): gte: Optional[List] = Field(None, title="Gte") gt: Optional[List] = Field(None, title="Gt") @@ -2846,6 +2855,17 @@ class TicketCategory(str, Enum): ANOMALY_TICKET = "ANOMALY_TICKET" +class TicketNotificationRequest(BaseModel): + entity_type: Optional[NotificationEntityType] = "ticket" + event_type: Optional[NotificationType] = "ticket_processed" + id: List[UUID] = Field(..., title="Id") + tenant_id: UUID = Field(..., title="Tenant Id") + + +class TicketNotificationResponse(BaseModel): + pass + + class TicketState(str, Enum): OPEN = "OPEN" CLOSED = "CLOSED" diff --git a/onelens_backend_client_v2/rpc/__init__.py b/onelens_backend_client_v2/rpc/__init__.py index 763e2d52..a42bbf91 100644 --- a/onelens_backend_client_v2/rpc/__init__.py +++ b/onelens_backend_client_v2/rpc/__init__.py @@ -130,6 +130,10 @@ TenantTicketActivityServiceRpcHandler, ) +from onelens_backend_client_v2.rpc.notification_service_rpc_handler import ( + NotificationServiceRpcHandler, +) + __all__ = [ "TenantVerifyServiceRpcHandler", @@ -165,4 +169,5 @@ "MetricsControlServiceRpcHandler", "TenantTicketAuditServiceRpcHandler", "TenantTicketActivityServiceRpcHandler", + "NotificationServiceRpcHandler", ] diff --git a/onelens_backend_client_v2/rpc/notification_service_rpc_handler.py b/onelens_backend_client_v2/rpc/notification_service_rpc_handler.py new file mode 100644 index 00000000..82db8f6e --- /dev/null +++ b/onelens_backend_client_v2/rpc/notification_service_rpc_handler.py @@ -0,0 +1,144 @@ +# NotificationServiceRpcHandler 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 TicketNotificationRequest + + +from onelens_backend_client_v2.models import TicketNotificationResponse + + +from onelens_backend_client_v2.api_client import ApiClient, RequestSerialized + + +class NotificationServiceRpcHandler: + """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 send_ticket_notification( + self, + request: TicketNotificationRequest, + _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, + ) -> TicketNotificationResponse: + """Send Ticket Notification + + + + :param request: (required) + :type request: TicketNotificationRequest + :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._send_ticket_notification_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": "TicketNotificationResponse", + "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 _send_ticket_notification_serialize( + self, + request: TicketNotificationRequest, + _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/notification_service/send_ticket_notification", + 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, + )