fastapi_15588

{"status": "ok", "stdout": "fastapi/applications.py:4576:        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/).\nfastapi/applications.py:4596:        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/#alternative-events-deprecated).\nfastapi/routing.py:64:    EventSourceResponse,\nfastapi/routing.py:65:    ServerSentEvent,\nfastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\nfastapi/routing.py:497:                # Generator endpoint: stream as Server-Sent Events\nfastapi/routing.py:501:                    if isinstance(item, ServerSentEvent):\nfastapi/routing.py:504:                        # For ServerSentEvent items we skip stream_item_field\nfastapi/routing.py:856:                    # EventSourceResponse (SSE).\nfastapi/routing.py:857:                    # ServerSentEvent is excluded: it's a transport\nfastapi/routing.py:862:                        or lenient_issubclass(response_class, EventSourceResponse)\nfastapi/routing.py:863:                    ) and not lenient_issubclass(stream_item, ServerSentEvent):\nfastapi/routing.py:971:            response_class, EventSourceResponse\nfastapi/routing.py:4929:        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/).\nfastapi/routing.py:4949:        [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/#alternative-events-deprecated).\nfastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\nfastapi/sse.py:8:# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\nfastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\nfastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\nfastapi/sse.py:24:    to enable Server Sent Events (SSE) responses.\n", "stderr": "", "exit_code": 0}

{"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": "        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            \"\"\"\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   

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

{"status": "ok", "stdout": "event: line1\nline2\n\n\n", "stderr": "", "exit_code": 0}

{"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_newlines(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": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

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

{"status": "ok", "stdout": "    data: Annotated[\n        Any,\n        Doc(\n            \"\"\"\n            The event payload.\n\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", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "        ),\n    ] = None\n    event: Annotated[\n        str | None,\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        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", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

{"status": "ok", "stdout": "    ] = None$\n    event: Annotated[$\n        str | None,$\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        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", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

{"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_newlines(v, \"event\")),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "is_truncated": false}

{"status": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

{"status": "ok", "stdout": "            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        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", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

{"status": "ok", "stdout": "    id: Annotated[$\n        str | None,$\n        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", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileEditError", "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context."}

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

{"status": "ok", "stdout": "Caught expected error for event: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE 'event' must not contain newlines [type=value_error, input_value='line1\\nline2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nCaught expected error for id: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newlines [type=value_error, input_value='line1\\nline2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nCaught expected error for id null: 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", "patch_size": 1147, "files_changed": 1}