fastapi_15588

{"status": "ok", "stdout": "./fastapi/.agents/skills/fastapi/references/streaming.md:16:To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.\n./fastapi/.agents/skills/fastapi/references/streaming.md:24:from fastapi.sse import EventSourceResponse\n./fastapi/.agents/skills/fastapi/references/streaming.md:35:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./fastapi/.agents/skills/fastapi/references/streaming.md:47:from fastapi.sse import EventSourceResponse, ServerSentEvent\n./fastapi/.agents/skills/fastapi/references/streaming.md:52:@app.get(\"/events\", response_class=EventSourceResponse)\n./fastapi/.agents/skills/fastapi/SKILL.md:333:See [the streaming reference](references/streaming.md) for JSON Lines, Server-Sent Events (`EventSourceResponse`, `ServerSentEvent`), and byte streaming (`StreamingResponse`) patterns.\n./fastapi/routing.py:64:    EventSourceResponse,\n./fastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\n./fastapi/routing.py:856:                    # EventSourceResponse (SSE).\n./fastapi/routing.py:862:                        or lenient_issubclass(response_class, EventSourceResponse)\n./fastapi/routing.py:971:            response_class, EventSourceResponse\n./fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\n./fastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\n./fastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n./fastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n./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-anno

{"status": "ok", "filepath": "fastapi/sse.py", "content": "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` 

{"status": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"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": "        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\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": "36:def _check_id_no_null(v: str | None) -> str | None:\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "     1\tfrom typing import Annotated, Any\n     2\t\n     3\tfrom annotated_doc import Doc\n     4\tfrom pydantic import AfterValidator, BaseModel, Field, model_validator\n     5\tfrom starlette.responses import StreamingResponse\n     6\t\n     7\t# Canonical SSE event schema matching the OpenAPI 3.2 spec\n     8\t# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n     9\t_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    10\t    \"type\": \"object\",\n    11\t    \"properties\": {\n    12\t        \"data\": {\"type\": \"string\"},\n    13\t        \"event\": {\"type\": \"string\"},\n    14\t        \"id\": {\"type\": \"string\"},\n    15\t        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    16\t    },\n    17\t}\n    18\t\n    19\t\n    20\tclass EventSourceResponse(StreamingResponse):\n    21\t    \"\"\"Streaming response with `text/event-stream` media type.\n    22\t\n    23\t    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    24\t    to enable Server Sent Events (SSE) responses.\n    25\t\n    26\t    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    27\t    with protocols like MCP that stream SSE over `POST`.\n    28\t\n    29\t    The actual encoding logic lives in the FastAPI routing layer. This class\n    30\t    serves mainly as a marker and sets the correct `Content-Type`.\n    31\t    \"\"\"\n    32\t\n    33\t    media_type = \"text/event-stream\"\n    34\t\n    35\t\n    36\tdef _check_id_no_null(v: str | None) -> str | None:\n    37\t    if v is not None and \"\\0\" in v:\n    38\t        raise ValueError(\"SSE 'id' must not contain null characters\")\n    39\t    return v\n    40\t\n    41\t\n    42\tclass ServerSentEvent(BaseModel):\n    43\t    \"\"\"Represents a single Server-Sent Event.\n    44\t\n    45\t    When `yield`ed from a *path operation function* that uses\n    46\t    `response_class=EventSourceResponse`, each 

{"status": "ok", "stdout": "            \"\"\"\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            \"\"\"\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\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:

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/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 2.87s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "tests/test_sse_validation.py", "size": 1083}

{"status": "error", "error_type": "CommandError", "error_message": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 4 items\n\ntests/test_sse_validation.py FF..                                        [100%]\n\n=================================== FAILURES ===================================\n________________________ test_sse_invalid_event_newline ________________________\n\n    def test_sse_invalid_event_newline():\n>       with pytest.raises(ValueError) as excinfo:\n             ^^^^^^^^^^^^^^^^^^^^^^^^^\nE       Failed: DID NOT RAISE ValueError\n\ntests/test_sse_validation.py:15: Failed\n_________________________ test_sse_invalid_id_newline __________________________\n\n    def test_sse_invalid_id_newline():\n>       with pytest.raises(ValueError) as excinfo:\n             ^^^^^^^^^^^^^^^^^^^^^^^^^\nE       Failed: DID NOT RAISE ValueError\n\ntests/test_sse_validation.py:22: Failed\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/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=========================== short test summary info ============================\nFAILED tests/test_sse_validation.py::test_sse_invalid_event_newli

{"status": "ok", "filepath": "fastapi/sse.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -33,6 +33,12 @@\n     media_type = \"text/event-stream\"\n \n \n+def _check_sse_no_newline(v: str | None, field_name: str) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(f\"SSE '{field_name}' must not contain newlines\")\n+    return v\n+\n+\n def _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", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/sse.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -92,6 +92,7 @@\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(lambda v: _check_sse_no_newline(v, \"event\")),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/sse.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -104,6 +104,7 @@\n     ] = None\n     id: Annotated[\n         str | None,\n+        AfterValidator(lambda v: _check_sse_no_newline(v, \"id\")),\n         AfterValidator(_check_id_no_null),\n         Doc(\n             \"\"\"\n", "is_truncated": false}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 4 items\n\ntests/test_sse_validation.py ....                                        [100%]\n\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/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======================== 4 passed, 1 warning in 17.67s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/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 2.12s =========================\n", "stderr": "", "exit_code": 0}

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

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