fastapi_15588

{"status": "ok", "stdout": "fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\nfastapi/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/openapi/docs.py:381:            document.addEventListener('DOMContentLoaded', function () {\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.\nfastapi/sse.py:42:class ServerSentEvent(BaseModel):\nfastapi/sse.py:43:    \"\"\"Represents a single Server-Sent Event.\nfastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\nfastapi/sse.py:93:            Maps to `addEventListener(event, ...)` on the browser. When omitted,\nfastapi/sse.py:105:            The browser sends this value back as the `Last-Event-ID` header on\nfastapi/sse.py:129:            `EventSource` clients. Useful for keep-alive pings to prevent\nfastapi/sse.py:136:    def _check_data_exclusive(self) -> \"ServerSentEvent\":\nfastapi/sse.py:140:                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\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\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "\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    For example, `data=\"hello\"` produces `data: \"hello\"` on the wire (with\n    quotes).\n    \"\"\"\n\n    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        ),\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        AfterVali

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

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

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

{"status": "ok", "stdout": "36:def _check_id_no_null(v: str | None) -> str | None:\n100:        AfterValidator(_check_id_no_null),\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "Failed: event with newline should have raised ValidationError\nFailed: id with newline should have raised ValidationError\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "Result: b'event: line1\\nline2\\ndata: hello\\n\\n'\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "Success: caught expected ValidationError for event: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE field 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\nFailed: id with newline should have raised ValidationError\nSuccess: caught expected ValidationError 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\nSuccess: valid event passed\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "Failed: id with newline should have raised ValidationError\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\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\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 strings** are JSON-serialized.\n\n    For example, `data=\"hello\"` produces `data: \"hello\"` on the wire (with\n    quotes).\n    \"\"\"\n\n    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 `da

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

{"status": "ok", "stdout": "Success: caught expected ValidationError for id: 1 validation error for ServerSentEvent\nid\n  Value error, SSE field 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\nSuccess: caught expected ValidationError 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", "stdout": "Success: caught expected ValidationError for event: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE field 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\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..a401847b2 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -33,6 +33,12 @@ class EventSourceResponse(StreamingResponse):\n     media_type = \"text/event-stream\"\n \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 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@@ -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@@ -97,6 +104,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     id: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines),\n         AfterValidator(_check_id_no_null),\n         Doc(\n             \"\"\"\n@@ -219,4 +227,4 @@ KEEPALIVE_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+_PING_INTERVAL: float = 15.0\n\\ No newline at end of file\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..5c0b80229 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -33,6 +33,12 @@ class EventSourceResponse(StreamingResponse):\n     media_type = \"text/event-stream\"\n \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 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@@ -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@@ -97,6 +104,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     id: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines),\n         AfterValidator(_check_id_no_null),\n         Doc(\n             \"\"\"\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": 1016, "files_changed": 1}