typing: Migrate to typed_endpoint.

This commit is contained in:
Kenneth Rodrigues
2024-06-06 17:35:03 +05:30
committed by Tim Abbott
parent 56db3f3d12
commit 6364010df5
2 changed files with 11 additions and 16 deletions

View File

@@ -117,7 +117,7 @@ class TypingValidateStreamIdTopicArgumentsTest(ZulipTestCase):
"/api/v1/typing",
{"type": "stream", "op": "start", "stream_id": "invalid", "topic": "test"},
)
self.assert_json_error(result, 'Argument "stream_id" is not valid JSON.')
self.assert_json_error(result, "stream_id is not valid JSON")
def test_includes_stream_id_but_not_topic(self) -> None:
sender = self.example_user("hamlet")

View File

@@ -2,32 +2,27 @@ from typing import List, Optional
from django.http import HttpRequest, HttpResponse
from django.utils.translation import gettext as _
from pydantic import Json
from typing_extensions import Annotated, Literal
from zerver.actions.typing import check_send_typing_notification, do_send_stream_typing_notification
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.streams import access_stream_by_id, access_stream_for_send_message
from zerver.lib.validator import check_int, check_list, check_string_in
from zerver.lib.typed_endpoint import ApiParamConfig, OptionalTopic, typed_endpoint
from zerver.models import UserProfile
VALID_OPERATOR_TYPES = ["start", "stop"]
VALID_RECIPIENT_TYPES = ["direct", "stream", "channel"]
@has_request_variables
@typed_endpoint
def send_notification_backend(
request: HttpRequest,
user_profile: UserProfile,
req_type: str = REQ(
"type", str_validator=check_string_in(VALID_RECIPIENT_TYPES), default="direct"
),
operator: str = REQ("op", str_validator=check_string_in(VALID_OPERATOR_TYPES)),
notification_to: Optional[List[int]] = REQ(
"to", json_validator=check_list(check_int), default=None
),
stream_id: Optional[int] = REQ(json_validator=check_int, default=None),
topic: Optional[str] = REQ("topic", default=None),
*,
req_type: Annotated[Literal["direct", "stream", "channel"], ApiParamConfig("type")] = "direct",
operator: Annotated[Literal["start", "stop"], ApiParamConfig("op")],
notification_to: Annotated[Json[Optional[List[int]]], ApiParamConfig("to")] = None,
stream_id: Json[Optional[int]] = None,
topic: OptionalTopic = None,
) -> HttpResponse:
recipient_type_name = req_type
if recipient_type_name == "channel":