⚡️ Speed up method EventSource._get_charset by 82%

[codeflash-ai]

📄 82% (0.82x) speedup for EventSource._get_charset in skyvern/client/core/http_sse/_api.py

⏱️ Runtime : 472 microseconds → 260 microseconds (best of 234 runs)

📝 Explanation and details

The optimized code achieves an 81% speedup by replacing expensive regex operations and string encode/decode validation with faster manual string parsing and codecs.lookup().

Key Optimizations:

1. Eliminates expensive regex: The original uses re.search(r"charset=([^;\s]+)", content_type, re.IGNORECASE) which compiles and executes a regex pattern. The optimized version uses simple string operations - content_type.lower().find("charset=") followed by manual character-by-character parsing to find the charset value boundary.

2. Faster charset validation: Instead of "test".encode(charset).decode(charset) which performs actual string encoding/decoding operations, the optimized code uses codecs.lookup(charset) which only validates that the charset name exists in Python's codec registry without performing expensive encoding operations.

3. Fast path for common case: When no charset is present (27% of test cases based on profiler data), the optimized version immediately returns "utf-8" after a single find() operation, avoiding all regex processing.

4. Manual boundary detection: The optimized code manually walks through characters to find where the charset value ends (at ;, whitespace, or end of string), which is faster than regex capture groups for simple parsing.

Performance Impact by Test Case:

• No charset cases: 67-99% faster (most common scenario)
• Large headers: 242-348% faster due to avoiding regex on long strings
• Standard charset cases: 35-51% faster across all valid charsets
• Invalid charset cases: 23-35% faster due to faster validation

The optimization is particularly effective for server-sent events (SSE) parsing where this function may be called frequently during streaming operations, making the cumulative performance gain significant for high-throughput applications.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 127 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import re

import httpx
# imports
import pytest
from skyvern.client.core.http_sse._api import EventSource

# unit tests

# -------- BASIC TEST CASES --------

def test_charset_utf8_explicit():
    """Test Content-Type header with explicit utf-8 charset."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.34μs -> 3.02μs (43.9% faster)

def test_charset_utf8_uppercase():
    """Test Content-Type header with uppercase UTF-8."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=UTF-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.09μs -> 2.79μs (46.5% faster)

def test_charset_iso_8859_1():
    """Test Content-Type header with ISO-8859-1 charset."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=ISO-8859-1"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.23μs -> 3.13μs (35.0% faster)

def test_charset_quoted():
    """Test Content-Type header with quoted charset value."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=\"utf-8\""})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.05μs -> 2.82μs (43.8% faster)

def test_charset_single_quoted():
    """Test Content-Type header with single quoted charset value."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset='utf-8'"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.99μs -> 2.85μs (40.0% faster)

def test_no_charset_in_header():
    """Test Content-Type header without charset (should default to utf-8)."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 2.92μs -> 1.61μs (81.2% faster)

def test_no_content_type_header():
    """Test missing Content-Type header (should default to utf-8)."""
    resp = httpx.Response(200, headers={})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.25μs -> 2.04μs (59.4% faster)

def test_charset_with_trailing_semicolon():
    """Test charset parameter with a trailing semicolon."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=utf-8;"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.98μs -> 2.95μs (35.0% faster)

# -------- EDGE TEST CASES --------

def test_charset_invalid_encoding():
    """Test Content-Type header with an invalid charset (should fallback to utf-8)."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=notarealcharset"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.89μs -> 4.47μs (31.7% faster)

def test_charset_with_spaces():
    """Test charset parameter with extra spaces."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=  utf-8  "})
    es = EventSource(resp)
    # The regex doesn't trim spaces inside the value, so this will fail to match and fallback to utf-8
    codeflash_output = es._get_charset() # 3.31μs -> 3.30μs (0.546% faster)

def test_charset_with_equal_sign_in_value():
    """Test charset parameter where value contains an equal sign (should parse only up to semicolon or whitespace)."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=utf-8=extra"})
    es = EventSource(resp)
    # "utf-8=extra" is not a valid encoding, should fallback
    codeflash_output = es._get_charset() # 5.63μs -> 4.37μs (28.9% faster)

def test_charset_with_multiple_parameters():
    """Test Content-Type header with multiple parameters after charset."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=utf-8; foo=bar"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.90μs -> 2.83μs (38.0% faster)

def test_charset_case_insensitivity():
    """Test charset parameter with mixed case (should be parsed case-insensitively)."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; CHARSET=iso-8859-1"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.95μs -> 3.02μs (30.8% faster)

def test_charset_with_leading_trailing_whitespace():
    """Test charset parameter with leading and trailing whitespace inside quotes."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset='  utf-8  '"})
    es = EventSource(resp)
    # The .strip() only removes quotes, not inner whitespace, so this will fail to match and fallback
    codeflash_output = es._get_charset() # 5.11μs -> 3.43μs (49.0% faster)

def test_charset_parameter_not_last():
    """Test charset parameter not being the last parameter."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; foo=bar; charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.85μs -> 2.79μs (37.6% faster)

def test_charset_with_tab_and_newline():
    """Test charset parameter with tab and newline characters."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream;\ncharset=utf-8\t"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.84μs -> 2.79μs (37.5% faster)

def test_content_type_header_empty_string():
    """Test Content-Type header with an empty string."""
    resp = httpx.Response(200, headers={"Content-Type": ""})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 2.70μs -> 1.61μs (67.3% faster)

def test_charset_with_non_ascii_encoding():
    """Test Content-Type header with a valid non-UTF encoding."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=cp1252"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 6.71μs -> 3.04μs (121% faster)

def test_charset_with_invalid_unicode_encoding():
    """Test Content-Type header with an invalid Unicode encoding (should fallback)."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=unicode-unknown"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.74μs -> 4.66μs (23.2% faster)

def test_charset_with_tricky_regex():
    """Test Content-Type header with tricky charset parameter placement."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; foo=bar; charset=utf-8; baz=qux"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.94μs -> 2.92μs (35.1% faster)

def test_charset_with_double_quotes_and_semicolon():
    """Test charset parameter with double quotes and semicolon inside quotes."""
    resp = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=\"utf-8;\""})
    es = EventSource(resp)
    # "utf-8;" is not a valid encoding, so fallback
    codeflash_output = es._get_charset() # 3.91μs -> 2.92μs (33.9% faster)

def test_many_headers_with_valid_charset():
    """Test with a large number of headers and a valid charset."""
    headers = {f"X-Test-{i}": "foo" for i in range(500)}
    headers["Content-Type"] = "text/event-stream; charset=utf-8"
    resp = httpx.Response(200, headers=headers)
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 13.9μs -> 12.3μs (13.3% faster)

def test_many_headers_without_content_type():
    """Test with a large number of headers and no Content-Type header."""
    headers = {f"X-Test-{i}": "foo" for i in range(800)}
    resp = httpx.Response(200, headers=headers)
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 16.7μs -> 15.1μs (10.3% faster)

def test_long_content_type_header():
    """Test with a very long Content-Type header and charset at the end."""
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(900)])
    content_type = f"text/event-stream; {long_params}; charset=utf-8"
    resp = httpx.Response(200, headers={"Content-Type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 57.2μs -> 12.7μs (348% faster)

def test_long_content_type_header_invalid_charset():
    """Test with a very long Content-Type header and invalid charset at the end."""
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(900)])
    content_type = f"text/event-stream; {long_params}; charset=notarealcharset"
    resp = httpx.Response(200, headers={"Content-Type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 59.1μs -> 14.0μs (323% faster)

def test_charset_within_large_header_value():
    """Test where charset is embedded in a large header value."""
    # Place charset in the middle of a long string
    pre = "; ".join([f"foo{i}=bar{i}" for i in range(400)])
    post = "; ".join([f"baz{i}=qux{i}" for i in range(400)])
    content_type = f"text/event-stream; {pre}; charset=utf-8; {post}"
    resp = httpx.Response(200, headers={"Content-Type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 27.5μs -> 9.30μs (196% faster)

# -------- ADDITIONAL EDGE CASES --------

@pytest.mark.parametrize(
    "header_value,expected",
    [
        ("text/event-stream; charset= utf-8", "utf-8"),  # leading space in value, regex will not match
        ("text/event-stream; charset=utf-8 ", "utf-8"),  # trailing space, regex will match
        ("text/event-stream; charset=", "utf-8"),        # empty value, regex will match but encode fails
        ("text/event-stream; charset=;", "utf-8"),       # value is semicolon, not a valid encoding
        ("text/event-stream; charset=\"utf-8", "utf-8"), # unclosed quote
        ("text/event-stream; charset=utf-8; charset=ISO-8859-1", "utf-8"), # first match wins
    ]
)
def test_various_tricky_headers(header_value, expected):
    """Test a variety of tricky header values."""
    resp = httpx.Response(200, headers={"Content-Type": header_value})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 22.0μs -> 18.9μs (16.8% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import re

import httpx
# imports
import pytest
from skyvern.client.core.http_sse._api import EventSource

# unit tests

# -------------------------
# Basic Test Cases
# -------------------------

def test_charset_explicit_utf8():
    # Standard header with explicit UTF-8
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=utf-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.38μs -> 3.10μs (41.4% faster)

def test_charset_explicit_utf16():
    # Standard header with explicit UTF-16
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=utf-16"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.42μs -> 2.93μs (51.2% faster)

def test_charset_explicit_iso8859_1():
    # Standard header with ISO-8859-1
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=ISO-8859-1"})
    es = EventSource(response)

def test_charset_explicit_uppercase():
    # Charset parameter in uppercase
    response = httpx.Response(200, headers={"content-type": "text/event-stream; CHARSET=UTF-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.16μs -> 2.89μs (44.0% faster)

def test_charset_with_quotes():
    # Charset value surrounded by quotes
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=\"utf-8\""})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.22μs -> 3.02μs (39.9% faster)

def test_charset_with_single_quotes():
    # Charset value surrounded by single quotes
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset='utf-8'"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.13μs -> 2.90μs (42.2% faster)

def test_charset_with_spaces():
    # Charset parameter with spaces around equals
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset = utf-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.21μs -> 1.61μs (99.1% faster)

def test_charset_with_semicolon_after():
    # Charset parameter at end with semicolon
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=utf-8;"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.93μs -> 2.86μs (37.6% faster)

def test_no_charset_returns_utf8():
    # No charset parameter present
    response = httpx.Response(200, headers={"content-type": "text/event-stream"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.01μs -> 1.60μs (87.6% faster)

def test_no_content_type_header():
    # No content-type header at all
    response = httpx.Response(200, headers={})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.20μs -> 2.06μs (55.5% faster)

# -------------------------
# Edge Test Cases
# -------------------------

def test_charset_invalid_encoding():
    # Invalid encoding should fall back to utf-8
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=not-a-real-charset"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 6.51μs -> 4.92μs (32.3% faster)

def test_charset_empty_value():
    # Empty charset value should fall back to utf-8
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset="})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.13μs -> 3.23μs (3.04% slower)

def test_charset_trailing_garbage():
    # Charset parameter with trailing garbage
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=utf-8;garbage"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.12μs -> 2.88μs (43.1% faster)

def test_charset_with_extra_parameters():
    # Charset with additional parameters after
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=utf-8; foo=bar"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.98μs -> 2.77μs (43.4% faster)

def test_charset_with_multiple_parameters():
    # Multiple parameters, charset not first
    response = httpx.Response(200, headers={"content-type": "text/event-stream; foo=bar; charset=utf-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.89μs -> 2.76μs (41.0% faster)

def test_charset_with_multiple_charsets():
    # Multiple charset parameters, should pick the first valid one
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=not-a-real-charset; charset=utf-8"})
    es = EventSource(response)
    # The regex will match the first, which is invalid, so fallback to utf-8
    codeflash_output = es._get_charset() # 5.89μs -> 4.36μs (35.0% faster)

def test_charset_with_non_ascii_encoding():
    # Charset with non-ASCII encoding name
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=cp1252"})
    es = EventSource(response)

def test_charset_with_leading_trailing_whitespace():
    # Charset value with leading/trailing whitespace
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=  utf-8  "})
    es = EventSource(response)
    # Regex does not strip whitespace inside value, so this is invalid and should fallback
    codeflash_output = es._get_charset() # 3.41μs -> 3.34μs (2.19% faster)

def test_charset_with_tab_and_newline():
    # Charset parameter with tab/newline characters
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=\tutf-8\n"})
    es = EventSource(response)
    # Regex does not strip \t or \n, so this is invalid and should fallback
    codeflash_output = es._get_charset() # 3.22μs -> 3.16μs (1.64% faster)

def test_charset_case_insensitive():
    # Charset value in mixed case
    response = httpx.Response(200, headers={"content-type": "text/event-stream; charset=UtF-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.96μs -> 2.90μs (36.1% faster)

def test_charset_with_nonstandard_header_case():
    # Header name in nonstandard case
    response = httpx.Response(200, headers={"Content-Type": "text/event-stream; charset=utf-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.96μs -> 2.84μs (39.7% faster)

def test_content_type_with_multiple_semicolons():
    # Multiple semicolons in header
    response = httpx.Response(200, headers={"content-type": "text/event-stream;;; charset=utf-8;;"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.86μs -> 2.72μs (41.6% faster)

def test_content_type_with_no_parameters():
    # Content-Type header with no parameters at all
    response = httpx.Response(200, headers={"content-type": "text/event-stream"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 2.99μs -> 1.62μs (84.6% faster)

def test_content_type_with_only_charset():
    # Content-Type header is just charset
    response = httpx.Response(200, headers={"content-type": "charset=utf-8"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 3.82μs -> 2.80μs (36.5% faster)

def test_content_type_with_charset_and_no_value():
    # Content-Type header with charset and no value
    response = httpx.Response(200, headers={"content-type": "charset="})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 2.67μs -> 3.16μs (15.5% slower)

def test_content_type_with_invalid_header_format():
    # Content-Type header with invalid format
    response = httpx.Response(200, headers={"content-type": ";;;;;;"})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 2.61μs -> 1.56μs (67.1% faster)

def test_content_type_with_charset_equals_only():
    # Content-Type header with just 'charset='
    response = httpx.Response(200, headers={"content-type": "charset="})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 2.58μs -> 3.12μs (17.4% slower)

# -------------------------
# Large Scale Test Cases
# -------------------------

def test_large_number_of_headers_irrelevant():
    # Many irrelevant headers, but correct charset present
    headers = {f"x-header-{i}": "foo" for i in range(500)}
    headers["content-type"] = "text/event-stream; charset=utf-8"
    response = httpx.Response(200, headers=headers)
    es = EventSource(response)
    codeflash_output = es._get_charset() # 12.6μs -> 11.4μs (10.9% faster)

def test_large_content_type_header_with_charset_at_end():
    # Very long content-type header, charset at the end
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(500)])
    header_value = f"text/event-stream; {long_params}; charset=utf-8"
    response = httpx.Response(200, headers={"content-type": header_value})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 33.1μs -> 8.90μs (272% faster)

def test_large_content_type_header_with_invalid_charset():
    # Very long content-type header, invalid charset
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(500)])
    header_value = f"text/event-stream; {long_params}; charset=notarealcharset"
    response = httpx.Response(200, headers={"content-type": header_value})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 35.0μs -> 10.2μs (242% faster)

def test_large_content_type_header_with_charset_at_start():
    # Very long content-type header, charset at the start
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(500)])
    header_value = f"text/event-stream; charset=utf-8; {long_params}"
    response = httpx.Response(200, headers={"content-type": header_value})
    es = EventSource(response)
    codeflash_output = es._get_charset() # 4.63μs -> 5.53μs (16.3% slower)

def test_large_content_type_header_with_multiple_charsets():
    # Very long header, first charset invalid, second valid
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(500)])
    header_value = f"text/event-stream; charset=notarealcharset; {long_params}; charset=utf-8"
    response = httpx.Response(200, headers={"content-type": header_value})
    es = EventSource(response)
    # The regex will match the first, which is invalid, so fallback to utf-8
    codeflash_output = es._get_charset() # 6.04μs -> 6.95μs (13.1% slower)

def test_large_content_type_header_with_valid_non_utf8_charset():
    # Very long header, valid non-utf8 charset
    long_params = "; ".join([f"foo{i}=bar{i}" for i in range(500)])
    header_value = f"text/event-stream; {long_params}; charset=latin1"
    response = httpx.Response(200, headers={"content-type": header_value})
    es = EventSource(response)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-EventSource._get_charset-miobypa5 and push.