⚡️ Speed up method EventSource._get_charset by 40%

[codeflash-ai]

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

⏱️ Runtime : 721 microseconds → 515 microseconds (best of 231 runs)

📝 Explanation and details

The optimization replaces expensive regex operations with faster string operations for the common case. The key changes are:

What was optimized:

• Fast path for charset detection: Uses content_type.lower().find("charset=") instead of immediately calling re.search() with regex pattern matching
• Manual string parsing: When charset is found, manually parses the value using string slicing and character-by-character scanning instead of regex capture groups
• Regex fallback: Only uses the original regex approach when the fast path fails to find "charset="

Why it's faster:

• String .find() operations are significantly faster than regex compilation and matching (90%+ of execution time was spent in the encode/decode validation, but the regex overhead was still meaningful)
• Manual character scanning for delimiters (; \t\r\n) avoids regex overhead for simple parsing
• The optimization particularly shines with large headers where charset appears early - avoiding expensive regex scanning of the entire string

Performance characteristics:

• Best case: 430% faster for large headers with charset at the end, 50% faster for typical UTF-8 cases
• Worst case: 10-25% slower when no charset is present (falls back to regex after failed fast path)
• Most common cases (valid charset present): 30-50% faster

The optimization is especially valuable since this function likely runs on every SSE connection establishment, making even small improvements impactful for high-throughput applications.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

import re
from typing import Dict

import httpx  # used to construct Response objects
# imports
import pytest  # used for our unit tests
from skyvern.client.core.http_sse._api import EventSource

# unit tests

# Helper function to create a dummy httpx.Response with given headers
def make_response(headers: Dict[str, str]) -> httpx.Response:
    # httpx.Response requires status_code and content; headers can be passed directly
    return httpx.Response(status_code=200, content=b"", headers=headers)

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

def test_valid_charset_utf8():
    """Basic: Should extract UTF-8 charset correctly."""
    resp = make_response({"content-type": "text/event-stream; charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.94μs -> 2.62μs (50.1% faster)

def test_valid_charset_iso8859_1():
    """Basic: Should extract ISO-8859-1 charset correctly."""
    resp = make_response({"content-type": "text/event-stream; charset=iso-8859-1"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.29μs -> 3.11μs (38.0% faster)

def test_no_charset_header():
    """Basic: Should default to UTF-8 if no charset specified."""
    resp = make_response({"content-type": "text/event-stream"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.06μs -> 4.01μs (23.8% slower)

def test_no_content_type_header():
    """Basic: Should default to UTF-8 if content-type header missing."""
    resp = make_response({})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.10μs -> 3.57μs (13.3% slower)

def test_charset_with_quotes():
    """Basic: Should handle charset parameter with quotes."""
    resp = make_response({"content-type": "text/event-stream; charset=\"utf-8\""})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.10μs -> 3.15μs (30.0% faster)

def test_charset_with_single_quotes():
    """Basic: Should handle charset parameter with single quotes."""
    resp = make_response({"content-type": "text/event-stream; charset='utf-8'"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.12μs -> 2.86μs (44.2% faster)

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

def test_invalid_charset_fallback():
    """Edge: Should fallback to UTF-8 for unknown charset."""
    resp = make_response({"content-type": "text/event-stream; charset=notarealcharset"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.90μs -> 4.95μs (19.3% faster)

def test_charset_with_extra_spaces():
    """Edge: Should handle extra spaces around charset value."""
    resp = make_response({"content-type": "text/event-stream; charset=   utf-8   "})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.29μs -> 3.69μs (11.0% slower)

def test_charset_case_insensitive():
    """Edge: Should handle charset parameter case insensitively."""
    resp = make_response({"content-type": "text/event-stream; CHARSET=UTF-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.78μs -> 2.69μs (40.5% faster)

def test_multiple_parameters():
    """Edge: Should extract charset when other parameters are present."""
    resp = make_response({"content-type": "text/event-stream; foo=bar; charset=utf-8; baz=qux"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.94μs -> 2.81μs (39.9% faster)

def test_charset_at_start_of_header():
    """Edge: Should extract charset even if it's the first parameter."""
    resp = make_response({"content-type": "charset=utf-8; text/event-stream"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.83μs -> 2.83μs (35.4% faster)

def test_charset_with_semicolon_in_value():
    """Edge: Should not include trailing semicolon in charset value."""
    resp = make_response({"content-type": "text/event-stream; charset=utf-8;"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.83μs -> 2.83μs (35.2% faster)

def test_charset_with_trailing_garbage():
    """Edge: Should extract charset even with trailing garbage."""
    resp = make_response({"content-type": "text/event-stream; charset=utf-8;garbage"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.84μs -> 2.79μs (37.7% faster)

def test_charset_with_unicode_encoding():
    """Edge: Should handle valid but less common encoding names."""
    resp = make_response({"content-type": "text/event-stream; charset=cp1252"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 6.38μs -> 5.41μs (17.9% faster)

def test_charset_with_unicode_error():
    """Edge: Should fallback to UTF-8 if encoding causes UnicodeError."""
    # 'utf-7' is valid but may cause UnicodeError for some operations
    resp = make_response({"content-type": "text/event-stream; charset=utf-7"})
    es = EventSource(resp)
    # Should not raise, should return 'utf-7' because encode/decode of "test" works
    codeflash_output = es._get_charset() # 5.87μs -> 4.76μs (23.4% faster)

def test_charset_with_lookup_error():
    """Edge: Should fallback to UTF-8 if encoding is not recognized."""
    resp = make_response({"content-type": "text/event-stream; charset=foobar"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.37μs -> 4.32μs (24.3% faster)

def test_charset_with_empty_value():
    """Edge: Should fallback to UTF-8 if charset is empty."""
    resp = make_response({"content-type": "text/event-stream; charset="})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 2.97μs -> 3.60μs (17.4% slower)

def test_charset_with_only_charset_parameter():
    """Edge: Should extract charset even if no mimetype is present."""
    resp = make_response({"content-type": "charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.65μs -> 2.68μs (36.3% faster)

def test_charset_with_multiple_charset_parameters():
    """Edge: Should extract first charset parameter if multiple present."""
    resp = make_response({"content-type": "text/event-stream; charset=iso-8859-1; charset=utf-8"})
    es = EventSource(resp)
    # Should pick 'iso-8859-1' as first match
    codeflash_output = es._get_charset() # 4.21μs -> 3.27μs (28.8% faster)

def test_charset_with_mixed_case_encoding():
    """Edge: Should handle charset value with mixed case."""
    resp = make_response({"content-type": "text/event-stream; charset=UtF-8"})
    es = EventSource(resp)

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

def test_large_header_with_charset_at_end():
    """Large Scale: Should extract charset from a large header string."""
    # Create a large header with many parameters, charset at the end
    params = [f"param{i}=value{i}" for i in range(990)]
    params.append("charset=utf-8")
    content_type = "text/event-stream; " + "; ".join(params)
    resp = make_response({"content-type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 77.7μs -> 14.7μs (430% faster)

def test_large_header_with_charset_at_start():
    """Large Scale: Should extract charset from a large header string with charset at start."""
    params = ["charset=iso-8859-1"] + [f"param{i}=value{i}" for i in range(990)]
    content_type = "text/event-stream; " + "; ".join(params)
    resp = make_response({"content-type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.79μs -> 9.65μs (39.9% slower)

def test_large_header_with_invalid_charset():
    """Large Scale: Should fallback to UTF-8 if charset is invalid in a large header."""
    params = [f"param{i}=value{i}" for i in range(990)]
    params.append("charset=notarealcharset")
    content_type = "text/event-stream; " + "; ".join(params)
    resp = make_response({"content-type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 79.6μs -> 16.5μs (383% faster)

def test_large_header_no_charset():
    """Large Scale: Should default to UTF-8 if no charset in a large header."""
    params = [f"param{i}=value{i}" for i in range(999)]
    content_type = "text/event-stream; " + "; ".join(params)
    resp = make_response({"content-type": content_type})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 77.3μs -> 88.1μs (12.3% slower)

def test_many_different_charset_encodings():
    """Large Scale: Should correctly extract and validate various encodings."""
    # Test a selection of valid encodings
    charsets = ["utf-8", "iso-8859-1", "cp1252", "latin1", "utf-16"]
    for charset in charsets:
        resp = make_response({"content-type": f"text/event-stream; charset={charset}"})
        es = EventSource(resp)

def test_many_invalid_charset_encodings():
    """Large Scale: Should fallback to UTF-8 for many invalid encodings."""
    # Test a selection of invalid encodings
    invalid_charsets = [f"invalid{i}" for i in range(10)]
    for charset in invalid_charsets:
        resp = make_response({"content-type": f"text/event-stream; charset={charset}"})
        es = EventSource(resp)
        codeflash_output = es._get_charset() # 23.8μs -> 18.9μs (26.3% 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 make_response_with_headers(headers):
    """Helper to create an httpx.Response with custom headers."""
    # httpx.Response requires status_code and content, but we only care about headers
    return httpx.Response(200, headers=headers, content=b"")

def test_charset_utf8_explicit():
    # Content-Type header explicitly specifies UTF-8
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.60μs -> 2.57μs (39.9% faster)

def test_charset_iso8859_1():
    # Content-Type header specifies ISO-8859-1
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=iso-8859-1"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.87μs -> 2.94μs (31.6% faster)

def test_charset_uppercase():
    # Charset parameter is uppercase, should be case-insensitive
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=UTF-16"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.03μs -> 2.87μs (40.5% faster)

def test_charset_with_quotes():
    # Charset value is quoted
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=\"utf-8\""})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.71μs -> 2.84μs (30.3% faster)

def test_charset_with_single_quotes():
    # Charset value is single-quoted
    resp = make_response_with_headers({"content-type": "text/event-stream; charset='utf-8'"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.97μs -> 2.72μs (45.8% faster)

def test_no_charset_header():
    # Content-Type header present but no charset specified
    resp = make_response_with_headers({"content-type": "text/event-stream"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 2.95μs -> 3.89μs (24.1% slower)

def test_no_content_type_header():
    # No Content-Type header at all
    resp = make_response_with_headers({})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.22μs -> 3.54μs (8.88% slower)

def test_charset_with_trailing_semicolon():
    # Charset parameter ends with a semicolon
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8;"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.02μs -> 2.92μs (37.7% faster)

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

def test_charset_invalid():
    # Charset specified is not a valid Python encoding
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=invalid-charset"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.64μs -> 4.63μs (21.9% faster)

def test_charset_empty_value():
    # Charset parameter present but with empty value
    resp = make_response_with_headers({"content-type": "text/event-stream; charset="})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.22μs -> 3.63μs (11.4% slower)

def test_charset_spaces_around_equals():
    # Spaces around equals sign
    resp = make_response_with_headers({"content-type": "text/event-stream; charset = utf-8"})
    es = EventSource(resp)
    # The regex does not match if there are spaces around '='
    codeflash_output = es._get_charset() # 3.22μs -> 3.94μs (18.3% slower)

def test_charset_with_additional_parameters():
    # Additional parameters after charset
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8; foo=bar"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.98μs -> 2.95μs (35.0% faster)

def test_charset_with_non_ascii_encoding():
    # Charset is a valid non-ASCII encoding
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=cp1252"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 6.48μs -> 5.09μs (27.2% faster)

def test_charset_with_tab_and_newline():
    # Charset value contains tab/newline (should not happen, but test anyway)
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8\n"})
    es = EventSource(resp)
    # The regex will match up to the newline, so should still work
    codeflash_output = es._get_charset() # 3.94μs -> 2.76μs (42.8% faster)

def test_charset_with_multiple_charset_parameters():
    # Multiple charset parameters (should pick the first)
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=iso-8859-1; charset=utf-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.26μs -> 3.18μs (34.0% faster)

def test_charset_with_leading_and_trailing_whitespace():
    # Charset value with leading/trailing whitespace
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=   utf-8   "})
    es = EventSource(resp)
    # The regex will only match up to the first whitespace, so returns 'utf-8'
    codeflash_output = es._get_charset() # 3.35μs -> 3.73μs (10.3% slower)

def test_charset_with_mixed_case_content_type():
    # Content-Type header with mixed case
    resp = make_response_with_headers({"Content-Type": "TeXt/EvEnT-StReAm; ChArSeT=UtF-8"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 3.72μs -> 2.71μs (37.5% faster)

def test_charset_with_semicolon_in_value():
    # Charset value contains a semicolon (should not be included)
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8;"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 5.58μs -> 3.76μs (48.5% faster)

def test_charset_with_charset_in_middle():
    # Charset parameter is not last
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8; foo=bar"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.39μs -> 3.00μs (46.4% faster)

def test_charset_with_unsupported_but_valid_python_encoding():
    # Charset is a valid Python encoding but not commonly used in HTTP
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=latin1"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.33μs -> 3.10μs (39.5% faster)

def test_charset_with_charset_and_value_in_quotes_and_semicolon():
    # Charset value is quoted and followed by semicolon
    resp = make_response_with_headers({"content-type": "text/event-stream; charset=\"utf-8\";"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 4.25μs -> 3.11μs (36.9% faster)

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

@pytest.mark.parametrize("encoding", [
    "utf-8", "iso-8859-1", "utf-16", "cp1252", "latin1", "ascii", "utf-32"
])
def test_many_valid_encodings(encoding):
    # Test a variety of valid encodings
    resp = make_response_with_headers({"content-type": f"text/event-stream; charset={encoding}"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 31.5μs -> 23.0μs (36.6% faster)

def test_large_number_of_headers():
    # Test with a large number of unrelated headers
    headers = {f"X-Test-{i}": f"value-{i}" for i in range(500)}
    headers["content-type"] = "text/event-stream; charset=utf-8"
    resp = make_response_with_headers(headers)
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 12.7μs -> 11.1μs (14.6% faster)

def test_large_content_type_header():
    # Test with a very large Content-Type header containing charset
    long_junk = ";".join([f"foo{i}=bar{i}" for i in range(500)])
    header = f"text/event-stream; {long_junk}; charset=utf-8"
    resp = make_response_with_headers({"content-type": header})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 31.1μs -> 7.81μs (298% faster)

def test_large_content_type_header_no_charset():
    # Test with a very large Content-Type header without charset
    long_junk = ";".join([f"foo{i}=bar{i}" for i in range(500)])
    header = f"text/event-stream; {long_junk}"
    resp = make_response_with_headers({"content-type": header})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 30.1μs -> 35.1μs (14.4% slower)

def test_large_invalid_charset():
    # Test with a large invalid charset value
    invalid_charset = "x" * 500
    resp = make_response_with_headers({"content-type": f"text/event-stream; charset={invalid_charset}"})
    es = EventSource(resp)
    codeflash_output = es._get_charset() # 10.9μs -> 15.1μs (27.7% slower)

def test_many_content_type_variants():
    # Test with 100 different valid charsets (only a few are valid, rest fallback)
    for i in range(100):
        charset = f"utf-8" if i % 10 == 0 else f"invalid{i}"
        resp = make_response_with_headers({"content-type": f"text/event-stream; charset={charset}"})
        es = EventSource(resp)
        expected = "utf-8" if charset == "utf-8" else "utf-8"
        codeflash_output = es._get_charset() # 172μs -> 136μs (25.9% faster)

# ----------- Determinism Test -----------

def test_determinism():
    # The same header should always yield the same result
    resp1 = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8"})
    resp2 = make_response_with_headers({"content-type": "text/event-stream; charset=utf-8"})
    es1 = EventSource(resp1)
    es2 = EventSource(resp2)
    codeflash_output = es1._get_charset() # 3.23μs -> 2.13μs (52.0% faster)
# 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-mjaph18l and push.