🐛 Fix line splitting in format_sse_event to comply with SSE spec

Project: tiangolo/fastapi | Language: Python | Concepts: serialization, testing-strategy, api-design | Difficulty: mid
Link: https://github.com/fastapi/fastapi/pull/15515

The problem

The format_sse_event function in FastAPI was incorrectly splitting lines in event payloads using Python's str.splitlines(). This caused two main issues:

• Trailing newlines in data were dropped, so payloads ending with a newline were corrupted.
• Non-SSE line breaks (like Unicode line separators and vertical tabs) were incorrectly treated as line breaks, splitting the data in ways not allowed by the SSE spec.

Why it's non-trivial

A naive fix might just replace splitlines() with a simple split on \n, but this would miss the fact that SSE allows three types of line breaks (\n, \r\n, and \r). It also requires preserving trailing empty lines, which splitlines() does not do. Correctly handling all these cases without breaking compatibility or introducing subtle bugs requires careful normalization.

The approach

A new helper function _split_sse_lines was introduced to handle line splitting according to the SSE specification:

def _split_sse_lines(value: str) -> list[str]:
    # Split on SSE-spec line terminators only (\n, \r\n, \r), preserving
    # trailing empty strings.
    return value.replace("\r\n", "\n").replace("\r", "\n").split("\n")

What this does:

• Normalizes all \r\n and \r to \n, so all three SSE line breaks are handled.
• Splits on \n, which preserves trailing empty strings (unlike splitlines()).
• Ignores non-SSE line breaks (like Unicode line separators or vertical tabs), so they are not treated as line breaks.

This helper is used in both the comment and data_str branches of format_sse_event:

-        for line in comment.splitlines():
+        for line in _split_sse_lines(comment):
             lines.append(f": {line}")

and

-        for line in data_str.splitlines():
+        for line in _split_sse_lines(data_str):
             lines.append(f"data: {line}")

What this does:

• Ensures both comments and data are split using the SSE-correct logic.

Tests were added to cover all relevant edge cases:

@pytest.mark.parametrize(
    ("data", "expected_result"),
    [
        ("Hello\n", b"data: Hello\ndata: \n\n"),
        ("Hello\n\n", b"data: Hello\ndata: \ndata: \n\n"),
        ("\n", b"data: \ndata: \n\n"),
        ("Hello\r\nWorld", b"data: Hello\ndata: World\n\n"),
        ("Hello\rWorld", b"data: Hello\ndata: World\n\n"),
        ("A\u2028B", "data: A\u2028B\n\n".encode()),
        ("A\vB", b"data: A\x0bB\n\n"),
        ("", b"data: \n\n"),
    ],
)
def test_format_sse_event_splitlines_behavior_in_data(
    data: str, expected_result: bytes
) -> None:
    assert format_sse_event(data_str=data) == expected_result

What this does:

• Verifies correct handling of trailing newlines, CRLF and CR normalization, and that Unicode line breaks and vertical tabs are not split.
• Explicitly checks that an empty string emits a correct empty data line.

A similar test was added for comments:

def test_format_sse_event_splitlines_behavior_in_comment():
    assert format_sse_event(comment="hi\n") == b": hi\n: \n\n"

What this does:

• Ensures that comments with trailing newlines are handled correctly.

From the review thread

YuriiMotov:
I would shorten tests this way.
Also, we should replace \u2028B with its code (as it's done in suggested changes) - otherwise VSCode shows warning when I open this file (see in the details)

YuriiMotov:
Let's also add the test case with empty data string from PR https://github.com/fastapi/fastapi/pull/15618:

def test_format_sse_event_keeps_empty_data_line():
    payload = format_sse_event(data_str="")
    assert payload == b"data: \n\n"

Takeaways

• Always consult and strictly follow external data format specifications (like SSE) when handling serialization or parsing, rather than relying on language built-ins that may have broader or different semantics.
• When splitting strings for protocols, ensure your method preserves all required edge cases, such as trailing empty lines, to avoid silent data corruption.
• Comprehensive, parameterized tests that cover edge cases and regressions are essential for serialization logic, especially when dealing with protocol compliance.