mirror of
https://github.com/zulip/zulip.git
synced 2025-11-02 04:53:36 +00:00
This demonstrates some basic use cases of the Json[...] wrapper with @typed_endpoint. Along with this change we extend test_openapi so that schema checking based on function signatures will still work with this new decorator. Pydantic's TypeAdapter supports dumping the JSON schema of any given type, which is leveraged here to validate against our own OpenAPI definitions. Parts of the implementation will be covered in later commits as we migrate more functions to use @typed_endpoint. See also: https://docs.pydantic.dev/latest/api/type_adapter/#pydantic.type_adapter.TypeAdapter.json_schema For the OpenAPI schema, we preprocess it mostly the same way. For the parameter types though, we no longer need to use get_standardized_argument_type to normalize type annotation, because Pydantic dumps a JSON schema that is compliant with OpenAPI schema already, which makes it a lot convenient for us to compare the types with our OpenAPI definitions. Do note that there are some exceptions where our definitions do not match the generated one. For example, we use JSON to parse int and bool parameters, but we don't mark them to use "application/json" in our definitions.
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from typing import List
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
from pydantic import Json, StringConstraints
|
|
from typing_extensions import Annotated
|
|
|
|
from zerver.actions.alert_words import do_add_alert_words, do_remove_alert_words
|
|
from zerver.lib.alert_words import user_alert_words
|
|
from zerver.lib.response import json_success
|
|
from zerver.lib.typed_endpoint import typed_endpoint
|
|
from zerver.models import UserProfile
|
|
|
|
|
|
def list_alert_words(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
|
|
return json_success(request, data={"alert_words": user_alert_words(user_profile)})
|
|
|
|
|
|
def clean_alert_words(alert_words: List[str]) -> List[str]:
|
|
alert_words = [w.strip() for w in alert_words]
|
|
return [w for w in alert_words if w != ""]
|
|
|
|
|
|
@typed_endpoint
|
|
def add_alert_words(
|
|
request: HttpRequest,
|
|
user_profile: UserProfile,
|
|
*,
|
|
alert_words: Json[List[Annotated[str, StringConstraints(max_length=100)]]],
|
|
) -> HttpResponse:
|
|
do_add_alert_words(user_profile, clean_alert_words(alert_words))
|
|
return json_success(request, data={"alert_words": user_alert_words(user_profile)})
|
|
|
|
|
|
@typed_endpoint
|
|
def remove_alert_words(
|
|
request: HttpRequest,
|
|
user_profile: UserProfile,
|
|
*,
|
|
alert_words: Json[List[str]],
|
|
) -> HttpResponse:
|
|
do_remove_alert_words(user_profile, alert_words)
|
|
return json_success(request, data={"alert_words": user_alert_words(user_profile)})
|