diff --git a/src/sentry/apidocs/_check_response_annotation_matches_schema.py b/src/sentry/apidocs/_check_response_annotation_matches_schema.py index 335abb641297..e932c2b023d7 100644 --- a/src/sentry/apidocs/_check_response_annotation_matches_schema.py +++ b/src/sentry/apidocs/_check_response_annotation_matches_schema.py @@ -1,4 +1,8 @@ -"""Lint: assert decorator's declared `T`s match the method's `Response[T]` annotation. +"""Lint: enforce typed response annotations on PUBLIC endpoints. + +This module runs two checks against every HTTP method in an endpoint class. + +## Check 1 — decorator `T` ⊆ annotation `T`s (drift) For each method decorated with @extend_schema(responses={N: inline_sentry_response_serializer("Name", T_decl), ...}) @@ -8,8 +12,8 @@ -> Response[T_success] | Response[T_error_400] | ... this linter asserts that the *set* of `T`s declared by -`inline_sentry_response_serializer(...)` entries in the decorator equals the set of -`T`s in the annotation's union arms. Names are compared verbatim — no +`inline_sentry_response_serializer(...)` entries in the decorator is a subset of +the `T`s in the annotation's union arms. Names are compared verbatim — no normalization, no convention-based pairing. mypy enforces body-vs-annotation; this linter enforces decorator-vs-annotation. @@ -17,7 +21,7 @@ it, and broad `except APIException` catches would lose the linkage anyway. Skipped silently (no diagnostic): - - Plain `-> Response` annotations (the unmigrated state). + - Plain `-> Response` annotations (handled by Check 2 for PUBLIC). - Methods with no `@extend_schema` decorator. - Decorator entries that aren't `inline_sentry_response_serializer(...)` — direct serializer-class references (`MonitorSerializer`), @@ -27,10 +31,30 @@ `inline_sentry_response_serializer(...)` or wait for the generic `Serializer[T]` refactor. +## Check 2 — PUBLIC endpoints must have typed return annotations + +For each HTTP method on an endpoint class whose `publish_status` declares the +method as `ApiPublishStatus.PUBLIC`, the return annotation must be one of: + - `Response[T]` for a concrete TypedDict / type `T` + - a `|`-union of `Response[T_i]` arms (and optionally non-`Response` arms + for endpoints that legitimately return Django responses, e.g. + `Response[Foo] | StreamingHttpResponse`) + - a non-`Response` Django response type if the endpoint never returns a + DRF `Response` (e.g. `HttpResponseBase`, `StreamingHttpResponse`, + `FileResponse`) + +The following are diagnostics: + - **missing annotation** — PUBLIC methods must explicitly declare what + they return so mypy can verify the body and drf-spectacular can + document it. + - **bare `Response`** — `-> Response` without `[T]` opts the method out of + type checking entirely. PUBLIC endpoints are documented in OpenAPI; an + untyped body lets the runtime drift from the declared schema. + Invoke as: python -m sentry.apidocs._check_response_annotation_matches_schema [paths...] -Exits non-zero on any mismatch. +Exits non-zero on any diagnostic. """ from __future__ import annotations @@ -55,6 +79,22 @@ ) +_HTTP_METHODS = frozenset({"get", "post", "put", "patch", "delete", "head", "options"}) + +_PUBLIC_HOW_TO = ( + "Fix by replacing the annotation with one of:\n" + " -> Response[YourResponseTypedDict] # single 200 shape\n" + " -> Response[YourResponseTypedDict] | Response[DetailResponse]\n" + " # success + error shape\n" + " -> Response[None] # explicit empty body\n" + " -> HttpResponseBase # streaming/redirect/non-DRF\n" + "If the body shape isn't documented anywhere yet, define a local\n" + "TypedDict in the endpoint file (`class _FooResponse(TypedDict): ...`)\n" + "and reference it. Standard error shapes (`{'detail': str}`,\n" + "`{field: errors}`) live in `sentry.apidocs.response_types`." +) + + @dataclass(frozen=True) class Mismatch: path: Path @@ -75,6 +115,31 @@ def __str__(self) -> str: ) +@dataclass(frozen=True) +class PublicUntyped: + """A PUBLIC HTTP method that lacks a usable return annotation.""" + + path: Path + line: int + cls: str + method: str + reason: str # "missing" or "bare-Response" + + def __str__(self) -> str: + if self.reason == "missing": + problem = "is PUBLIC but has no return annotation" + else: + problem = "is PUBLIC but annotated with bare `Response` (no `[T]`)" + return ( + f"{self.path}:{self.line} {self.cls}.{self.method}: {problem}.\n" + "Why: PUBLIC endpoints are documented in the OpenAPI spec and\n" + "consumed by SDK generators and external clients. An untyped\n" + "return lets the runtime body drift from the declared schema\n" + "without anyone noticing (mypy can't verify what isn't typed).\n" + f"{_PUBLIC_HOW_TO}" + ) + + def _name_of(node: ast.expr) -> str: """Render `Foo`, `mod.Foo`, `Foo[T]` as a stable string for equality.""" if isinstance(node, ast.Name): @@ -188,6 +253,142 @@ def _iter_methods(tree: ast.Module) -> Iterator[tuple[str, ast.FunctionDef | ast yield node.name, item +def _iter_classes(tree: ast.Module) -> Iterator[ast.ClassDef]: + for node in tree.body: + if isinstance(node, ast.ClassDef): + yield node + + +def _publish_status(cls: ast.ClassDef) -> dict[str, str]: + """Return `{HTTP_METHOD: ApiPublishStatus_attr_name}` parsed from + `publish_status = {"GET": ApiPublishStatus.PUBLIC, ...}` on the class + body. Also handles the annotated form + `publish_status: dict[str, ApiPublishStatus] = {...}`. Returns `{}` + when the class has no such assignment or the value doesn't statically + match the expected literal-dict shape.""" + for item in cls.body: + value: ast.expr | None = None + if ( + isinstance(item, ast.Assign) + and len(item.targets) == 1 + and isinstance(item.targets[0], ast.Name) + and item.targets[0].id == "publish_status" + ): + value = item.value + elif ( + isinstance(item, ast.AnnAssign) + and isinstance(item.target, ast.Name) + and item.target.id == "publish_status" + and item.value is not None + ): + value = item.value + if not isinstance(value, ast.Dict): + continue + out: dict[str, str] = {} + for k, v in zip(value.keys, value.values): + if ( + isinstance(k, ast.Constant) + and isinstance(k.value, str) + and isinstance(v, ast.Attribute) + ): + out[k.value] = v.attr + return out + return {} + + +_UNION_NAMES = frozenset({"Union", "Optional"}) + + +def _is_union_subscript(node: ast.expr) -> ast.expr | None: + """If `node` is `Union[a, b, ...]`, `Optional[a]`, or the `typing.`-prefixed + forms, return the slice expression (a `Tuple` of arms for `Union`, or a + single arm for `Optional`/`Union[X]`). Otherwise return None. + + `Optional[X]` is `Union[X, None]` semantically — the `None` arm has no + way to encode a bare `Response`, so unwrapping just the `X` is enough.""" + if not isinstance(node, ast.Subscript): + return None + val = node.value + if isinstance(val, ast.Name) and val.id in _UNION_NAMES: + return node.slice + if isinstance(val, ast.Attribute) and val.attr in _UNION_NAMES: + return node.slice + return None + + +def _annotation_has_bare_response(returns: ast.expr) -> bool: + """Walk a return annotation's union arms; return True iff any arm is the + bare `Response` name (no `[T]` subscript). Handles both `X | Y` and + `Union[X, Y]` union forms.""" + pending: list[ast.expr] = [returns] + while pending: + node = pending.pop() + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + pending.append(node.left) + pending.append(node.right) + continue + union_slice = _is_union_subscript(node) + if union_slice is not None: + if isinstance(union_slice, ast.Tuple): + pending.extend(union_slice.elts) + else: + pending.append(union_slice) + continue + if isinstance(node, ast.Name) and node.id == "Response": + return True + if isinstance(node, ast.Attribute) and node.attr == "Response": + # `rest_framework.response.Response` — bare attribute form + return True + return False + + +def check_file_public_typed(path: Path) -> list[PublicUntyped]: + """Diagnose PUBLIC methods that lack a usable return annotation.""" + try: + source = path.read_text() + except OSError: + return [] + try: + tree = ast.parse(source) + except SyntaxError: + return [] + + out: list[PublicUntyped] = [] + for cls in _iter_classes(tree): + ps = _publish_status(cls) + if not ps: + continue + for item in cls.body: + if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if item.name not in _HTTP_METHODS: + continue + if ps.get(item.name.upper()) != "PUBLIC": + continue + if item.returns is None: + out.append( + PublicUntyped( + path=path, + line=item.lineno, + cls=cls.name, + method=item.name, + reason="missing", + ) + ) + continue + if _annotation_has_bare_response(item.returns): + out.append( + PublicUntyped( + path=path, + line=item.lineno, + cls=cls.name, + method=item.name, + reason="bare-Response", + ) + ) + return out + + def check_file(path: Path) -> list[Mismatch]: try: source = path.read_text() @@ -248,18 +449,32 @@ def iter_files(roots: Iterable[str]) -> Iterator[Path]: def main(argv: list[str]) -> int: roots = argv[1:] if len(argv) > 1 else list(DEFAULT_PATHS) all_mismatches: list[Mismatch] = [] + all_untyped: list[PublicUntyped] = [] for path in iter_files(roots): all_mismatches.extend(check_file(path)) + all_untyped.extend(check_file_public_typed(path)) + for m in all_mismatches: sys.stdout.write(f"{m}\n") + for u in all_untyped: + sys.stdout.write(f"{u}\n") + if all_mismatches: sys.stderr.write( - f"\n{len(all_mismatches)} mismatch(es) — every `T` declared by " - "`inline_sentry_response_serializer(...)` in `@extend_schema` must " - "appear in the `Response[T]` (or union) annotation. The annotation " - "MAY declare additional arms (e.g. local error TypedDicts) that the " - "decorator does not expose.\n", + f"\n{len(all_mismatches)} decorator/annotation mismatch(es) — every " + "`T` declared by `inline_sentry_response_serializer(...)` in " + "`@extend_schema` must appear in the `Response[T]` (or union) " + "annotation. The annotation MAY declare additional arms (e.g. " + "local error TypedDicts) that the decorator does not expose.\n", + ) + if all_untyped: + sys.stderr.write( + f"\n{len(all_untyped)} PUBLIC method(s) without a typed return " + "annotation. PUBLIC endpoints are documented in OpenAPI and must " + "declare the runtime response shape statically so mypy + the " + "OpenAPI generator can keep the spec honest.\n" ) + if all_mismatches or all_untyped: return 1 return 0 diff --git a/tests/sentry/apidocs/test_check_response_annotation_matches_schema.py b/tests/sentry/apidocs/test_check_response_annotation_matches_schema.py index 4a7d69ff518e..319ed38549f5 100644 --- a/tests/sentry/apidocs/test_check_response_annotation_matches_schema.py +++ b/tests/sentry/apidocs/test_check_response_annotation_matches_schema.py @@ -7,11 +7,23 @@ from sentry.apidocs._check_response_annotation_matches_schema import ( Mismatch, + PublicUntyped, check_file, + check_file_public_typed, main, ) +def _run_public(source: str) -> list[PublicUntyped]: + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(source) + path = Path(f.name) + try: + return check_file_public_typed(path) + finally: + path.unlink() + + def _run(source: str) -> list[Mismatch]: with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(source) @@ -512,3 +524,303 @@ def get(self) -> Response[FooResponse] | Response[DetailResponse]: # Critically: the linter passes NOT because it special-cases DetailResponse, # but because the subset rule accepts any extra annotation arm. assert _run(source) == [] + + +# --------------------------------------------------------------------------- +# PUBLIC-must-be-typed check +# --------------------------------------------------------------------------- + + +def test_public_bare_response_fires() -> None: + source = """ +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response: + return Response() +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].method == "get" + assert diags[0].reason == "bare-Response" + assert "PUBLIC but annotated with bare `Response`" in str(diags[0]) + assert "Fix by replacing the annotation with one of" in str(diags[0]) + + +def test_public_missing_annotation_fires() -> None: + source = """ +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"POST": ApiPublishStatus.PUBLIC} + def post(self): + return Response() +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].method == "post" + assert diags[0].reason == "missing" + assert "PUBLIC but has no return annotation" in str(diags[0]) + + +def test_public_response_t_passes() -> None: + source = """ +from typing import TypedDict +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response[FooResponse]: + return Response({"x": 1}) +""" + assert _run_public(source) == [] + + +def test_public_union_with_response_arms_passes() -> None: + source = """ +from typing import TypedDict +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class DetailResponse(TypedDict): + detail: str + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response[FooResponse] | Response[DetailResponse]: + return Response({"x": 1}) +""" + assert _run_public(source) == [] + + +def test_public_non_drf_response_annotation_passes() -> None: + """Endpoints that legitimately return a Django HttpResponse subclass + (e.g. StreamingHttpResponse, FileResponse, or `http_method_not_allowed`) + don't have a DRF `Response[T]` to type — the explicit non-`Response` + annotation is the documented return shape.""" + source = """ +from django.http.response import HttpResponseBase + +class FooEndpoint: + publish_status = {"PUT": ApiPublishStatus.PUBLIC} + def put(self) -> HttpResponseBase: + return self.http_method_not_allowed(None) +""" + assert _run_public(source) == [] + + +def test_public_response_t_with_streaming_arm_passes() -> None: + """Mixed unions of Response[T] arms with non-`Response` arms (e.g. the + `?download` paths that stream a binary body) are accepted — what matters + is that no bare `Response` arm appears.""" + source = """ +from typing import TypedDict +from django.http import StreamingHttpResponse +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response[FooResponse] | StreamingHttpResponse: + return Response({"x": 1}) +""" + assert _run_public(source) == [] + + +def test_private_bare_response_does_not_fire() -> None: + """Only PUBLIC methods are linted — PRIVATE/EXPERIMENTAL endpoints stay + in the unmigrated state without diagnostic.""" + source = """ +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PRIVATE} + def get(self) -> Response: + return Response() +""" + assert _run_public(source) == [] + + +def test_method_outside_publish_status_does_not_fire() -> None: + """`publish_status` only declares some methods; ones not in the dict + (e.g. helper methods that share an HTTP-method name on a non-endpoint + class) aren't linted.""" + source = """ +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response[dict]: + return Response({}) + def post(self) -> Response: + return Response() +""" + # `post` isn't in publish_status — skipped. + assert _run_public(source) == [] + + +def test_class_without_publish_status_skipped() -> None: + """Non-endpoint classes (helpers, mixins, validators) don't have a + `publish_status` dict — they're not checked.""" + source = """ +from rest_framework.response import Response + +class SomeMixin: + def get(self) -> Response: + return Response() +""" + assert _run_public(source) == [] + + +def test_public_publish_status_annotated_assign_detected() -> None: + """`publish_status: dict[str, ApiPublishStatus] = {...}` (annotated + assignment) is read the same as the plain-assignment form.""" + source = """ +from rest_framework.response import Response + +class FooEndpoint: + publish_status: dict[str, ApiPublishStatus] = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response: + return Response() +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].reason == "bare-Response" + + +def test_public_bare_response_in_typing_union_fires() -> None: + """`-> Union[Response, Response[Foo]]` is just as bad as `Response` + alone — the bare arm opts that path out of body-vs-schema verification.""" + source = """ +from typing import TypedDict, Union +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Union[Response, Response[FooResponse]]: + return Response({"x": 1}) +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].reason == "bare-Response" + + +def test_public_bare_response_in_optional_fires() -> None: + """`Optional[Response]` is `Union[Response, None]` — the bare `Response` + arm is still detected.""" + source = """ +from typing import Optional +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Optional[Response]: + return Response() +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].reason == "bare-Response" + + +def test_public_optional_of_typed_response_passes() -> None: + """`Optional[Response[T]]` is accepted — the `Response[T]` arm is typed + and the implicit `None` arm doesn't introduce a bare `Response`.""" + source = """ +from typing import TypedDict, Optional +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Optional[Response[FooResponse]]: + return Response({"x": 1}) +""" + assert _run_public(source) == [] + + +def test_public_typing_union_of_typed_response_passes() -> None: + """`Union[Response[A], Response[B]]` is equivalent to the `|` form and + is accepted.""" + source = """ +from typing import TypedDict, Union +from rest_framework.response import Response + +class FooResponse(TypedDict): + x: int + +class DetailResponse(TypedDict): + detail: str + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Union[Response[FooResponse], Response[DetailResponse]]: + return Response({"x": 1}) +""" + assert _run_public(source) == [] + + +def test_public_bare_response_under_future_annotations_fires() -> None: + """`from __future__ import annotations` (PEP 563) defers annotation + evaluation *at runtime* — `__annotations__` stores strings instead of + resolved types — but `ast.parse()` still produces full expression nodes + for them. The bare-`Response` check walks the AST, not runtime + `__annotations__`, so the future import is a no-op for this linter.""" + source = """ +from __future__ import annotations +from rest_framework.response import Response + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC} + def get(self) -> Response: + return Response() +""" + diags = _run_public(source) + assert len(diags) == 1 + assert diags[0].reason == "bare-Response" + + +def test_main_emits_both_diagnostics(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """The unified `main` runs both checks and exit-codes non-zero on either.""" + src = """ +from typing import TypedDict +from drf_spectacular.utils import extend_schema +from rest_framework.response import Response +from sentry.apidocs.utils import inline_sentry_response_serializer + +class FooResponse(TypedDict): + x: int + +class FooEndpoint: + publish_status = {"GET": ApiPublishStatus.PUBLIC, "POST": ApiPublishStatus.PUBLIC} + + @extend_schema( + responses={200: inline_sentry_response_serializer("Foo", FooResponse)}, + ) + def get(self) -> Response: + return Response({"x": 1}) + + def post(self) -> Response[FooResponse]: + return Response({"x": 1}) +""" + path = tmp_path / "endpoints.py" + path.write_text(src) + rc = main(["check", str(path)]) + assert rc == 1 + captured = capsys.readouterr() + # The PUBLIC-untyped diagnostic for `get` should appear. + assert "FooEndpoint.get: is PUBLIC but annotated with bare `Response`" in captured.out + # `post` is correctly typed → no diagnostic for it. + assert "FooEndpoint.post:" not in captured.out