fastapi_15588

{"status": "ok", "stdout": "./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./tests/test_sse.py:35:@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\n./tests/test_sse.py:40:@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:46:@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:51:@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\n./tests/test_sse.py:57:@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\n./tests/test_sse.py:65:@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\n./tests/test_sse.py:72:@app.get(\"/items/stream-string\", response_class=EventSourceResponse)\n./tests/test_sse.py:77:@app.post(\"/items/stream-post\", response_class=EventSourceResponse)\n./tests/test_sse.py:83:@app.get(\"/items/stream-raw\", response_class=EventSourceResponse)\n./tests/test_sse.py:93:@router.get(\"/events\", response_class=EventSourceResponse)\n./tests/test_sse.py:274:@keepalive_app.get(\"/slow-async\", response_class=EventSourceResponse)\n./tests/test_sse.py:283:@keepalive_app.get(\"/slow-sync\", response_class=EventSourceResponse)\n./docs/zh-hant/docs/tutorial/server-sent-events.md:38:\u8981\u5728 FastAPI \u4e2d\u4e32\u6d41 SSE\uff0c\u8acb\u5728\u4f60\u7684\u8def\u5f91\u64cd\u4f5c\u51fd\u5f0f\uff08path operation function\uff09\u4e2d\u4f7f\u7528 `yield`\uff0c\u4e26\u8a2d\u5b9a `response_class=EventSourceResponse`\u3002\n./docs/zh-hant/docs/tutorial/server-sent-events.md:40:\u5f9e `fastapi.sse` \u532f\u5165 `EventSourceResponse`\uff1a\n./docs/ja/docs/tutorial/server-sent-events.md:38:FastAPI \u3067 SSE \u3092\u30b9\u30c8\u30ea\u30fc\u30df\u30f3\u30b0\u3059\u308b\u306b\u306f\u3001*path operation \u95a2\u6570*\u3067 `yield` \u3092\u4f7f\u3044\u3001`response_class=EventSourceResponse` \u30

{"status": "ok", "stdout": "./fastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string

{"status": "ok", "stdout": "        AfterValidator(_check_id_no_null),\n        Doc(\n            \"\"\"\n            Optional event ID.\n\n            The browser sends this value back as the `Last-Event-ID` header on\n            automatic reconnection. **Must not contain null (`\\\\0`) characters.**\n            \"\"\"\n        ),\n    ] = None\n    retry: Annotated[\n        int | None,\n        Field(ge=0),\n        Doc(\n            \"\"\"\n            Optional reconnection time in **milliseconds**.\n\n            Tells the browser how long to wait before reconnecting after the\n            connection is lost. Must be a non-negative integer.\n            \"\"\"\n        ),\n    ] = None\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s).\n\n            Comment lines start with `:` in the SSE wire format and are ignored by\n            `EventSource` clients. Useful for keep-alive pings to prevent\n            proxy/load-balancer timeouts.\n            \"\"\"\n        ),\n    ] = None\n\n    @model_validator(mode=\"after\")\n    def _check_data_exclusive(self) -> \"ServerSentEvent\":\n        if self.data is not None and self.raw_data is not None:\n            raise ValueError(\n                \"Cannot set both 'data' and 'raw_data' on the same \"\n                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\n                \"or 'raw_data' for pre-formatted strings.\"\n            )\n        return self\n\n\ndef format_sse_event(\n    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event ID (`id:` field).\n            \"\"\"\n        ),\n    ] = None,\n    retry: Annotated[\n        int | None,\n        Doc(\n            \"\"\"\n            Optional reconnection time in milliseconds (`retry:` field).\n            \"\"\"\n        ),\n    ] = None,\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s) (`:` prefix).\n            \"\"\"\n        ),\n    ] = None,\n) -> bytes:\n    \"\"\"Build SSE wire-format bytes from **pre-serialized** data.\n\n    The result always ends with `\\n\\n` (the event terminator).\n    \"\"\"\n    lines: list[str] = []\n\n    if comment is not None:\n        for line in comment.splitlines():\n            lines.append(f\": {line}\")\n\n    if event is not None:\n        lines.append(f\"event: {event}\")\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "        lines.append(f\"event: {event}\")\n\n    if data_str is not None:\n        for line in data_str.splitlines():\n            lines.append(f\"data: {line}\")\n\n    if id is not None:\n        lines.append(f\"id: {id}\")\n\n    if retry is not None:\n        lines.append(f\"retry: {retry}\")\n\n    lines.append(\"\")\n    lines.append(\"\")\n    return \"\\n\".join(lines).encode(\"utf-8\")\n\n\n# Keep-alive comment, per the SSE spec recommendation\nKEEPALIVE_COMMENT = b\": ping\\n\\n\"\n\n# Seconds between keep-alive pings when a generator is idle.\n# Private but importable so tests can monkeypatch it.\n_PING_INTERVAL: float = 15.0\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "import asyncio\nimport time\nfrom collections.abc import AsyncIterable, Iterable\n\nimport fastapi.routing\nimport pytest\nfrom fastapi import APIRouter, FastAPI\nfrom fastapi.responses import EventSourceResponse\nfrom fastapi.sse import ServerSentEvent\nfrom fastapi.testclient import TestClient\nfrom pydantic import BaseModel\n\n\nclass Item(BaseModel):\n    name: str\n    description: str | None = None\n\n\nitems = [\n    Item(name=\"Plumbus\", description=\"A multi-purpose household device.\"),\n    Item(name=\"Portal Gun\", description=\"A portal opening device.\"),\n    Item(name=\"Meeseeks Box\", description=\"A box that summons a Meeseeks.\"),\n]\n\n\napp = FastAPI()\n\n\n@app.get(\"/items/stream\", response_class=EventSourceResponse)\nasync def sse_items() -> AsyncIterable[Item]:\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\ndef sse_items_sync() -> Iterable[Item]:\n    yield from items\n\n\n@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\nasync def sse_items_no_annotation():\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\ndef sse_items_sync_no_annotation():\n    yield from items\n\n\n@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\nasync def sse_items_dict():\n    for item in items:\n        yield {\"name\": item.name, \"description\": item.description}\n\n\n@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\nasync def sse_items_event():\n    yield ServerSentEvent(data=\"hello\", event=\"greeting\", id=\"1\")\n    yield ServerSentEvent(data={\"key\": \"value\"}, event=\"json-data\", id=\"2\")\n    yield ServerSentEvent(comment=\"just a comment\")\n    yield ServerSentEvent(data=\"retry-test\", retry=5000)\n\n\n@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\nasync def sse_items_mix

{"status": "ok", "stdout": "{'data': 'hello', 'raw_data': None, 'event': 'multi\\nline', 'id': None, 'retry': None, 'comment': None}\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "{'data': 'hello', 'raw_data': None, 'event': None, 'id': 'multi\\nline', 'retry': None, 'comment': None}\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "event: multi\nline\ndata: hello\n\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "data: hello\nid: multi\nline\n\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "Event multi-line allowed\nId multi-line allowed\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "class EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\ndef _check_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\n\" in v:\n        raise ValueError(\"SSE field must not contain newlines\")\n    return v\n\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain strings** are JSON-serialized.\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "\n            Can be any JSON-serializable value: a Pydantic model, dict, list,\n            string, number, etc. It is **always** serialized to JSON: strings\n            are quoted (`\"hello\"` becomes `data: \"hello\"` on the wire).\n\n            Mutually exclusive with `raw_data`.\n            \"\"\"\n        ),\n    ] = None\n    raw_data: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Raw string to send as the `data:` field **without** JSON encoding.\n\n            Use this when you need to send pre-formatted text, HTML fragments,\n            CSV lines, or any non-JSON payload. The string is placed directly\n            into the `data:` field as-is.\n\n            Mutually exclusive with `data`.\n            \"\"\"\n        ),\n    ] = None\n    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines),\n        Doc(\n            \"\"\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \"\"\"\n        ),\n    ] = None\n    id: Annotated[\n        str | None,\n        AfterValidator(_check_id_no_null),\n        AfterValidator(_check_no_newlines),\n        Doc(\n            \"\"\"\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "Event multi-line caught: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE field must not contain newlines [type=value_error, input_value='multi\\nline', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nId multi-line caught: 1 validation error for ServerSentEvent\nid\n  Value error, SSE field must not contain newlines [type=value_error, input_value='multi\\nline', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nId null char caught: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain null characters [type=value_error, input_value='null\\x00char', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /private/workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n======================== 18 passed, 1 warning in 1.17s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..97ae49e47 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -38,6 +38,12 @@ def _check_id_no_null(v: str | None) -> str | None:\n         raise ValueError(\"SSE 'id' must not contain null characters\")\n     return v\n \n+def _check_no_newlines(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE field must not contain newlines\")\n+    return v\n+\n+\n \n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n@@ -86,6 +92,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name.\n@@ -98,6 +105,7 @@ class ServerSentEvent(BaseModel):\n     id: Annotated[\n         str | None,\n         AfterValidator(_check_id_no_null),\n+        AfterValidator(_check_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event ID.\n@@ -155,6 +163,7 @@ def format_sse_event(\n     ] = None,\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name (`event:` field).\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines),\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "format_sse_event multi-line allowed\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "./fastapi/sse.py:154:def format_sse_event(\n./fastapi/routing.py:66:    format_sse_event,\n./fastapi/routing.py:515:                        return format_sse_event(\n./fastapi/routing.py:525:                        return format_sse_event(\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "                def _serialize_sse_item(item: Any) -> bytes:\n                    if isinstance(item, ServerSentEvent):\n                        # User controls the event structure.\n                        # Serialize the data payload if present.\n                        # For ServerSentEvent items we skip stream_item_field\n                        # validation (the user may mix types intentionally).\n                        if item.raw_data is not None:\n                            data_str: str | None = item.raw_data\n                        elif item.data is not None:\n                            if hasattr(item.data, \"model_dump_json\"):\n                                data_str = item.data.model_dump_json()\n                            else:\n                                data_str = json.dumps(jsonable_encoder(item.data))\n                        else:\n                            data_str = None\n                        return format_sse_event(\n                            data_str=data_str,\n                            event=item.event,\n                            id=item.id,\n                            retry=item.retry,\n                            comment=item.comment,\n                        )\n                    else:\n                        # Plain object: validate + serialize via\n                        # stream_item_field (if set) and wrap in data field\n                        return format_sse_event(\n                            data_str=_serialize_data(item).decode(\"utf-8\")\n                        )\n\n                if dependant.is_async_gen_callable:\n                    sse_aiter: AsyncIterator[Any] = gen.__aiter__()\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /private/workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n======================== 18 passed, 1 warning in 1.13s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..dbb6f6fa0 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -38,6 +38,12 @@ def _check_id_no_null(v: str | None) -> str | None:\n         raise ValueError(\"SSE 'id' must not contain null characters\")\n     return v\n \n+def _check_no_newlines(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE field must not contain newlines\")\n+    return v\n+\n+\n \n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n@@ -86,6 +92,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name.\n@@ -98,6 +105,7 @@ class ServerSentEvent(BaseModel):\n     id: Annotated[\n         str | None,\n         AfterValidator(_check_id_no_null),\n+        AfterValidator(_check_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event ID.\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: submit_patch: command not found\n", "details": {"stdout": "", "stderr": "/bin/bash: submit_patch: command not found\n", "exit_code": 127}}

{"status": "ok", "patch_size": 1008, "files_changed": 1}