python/cpython · warning · TypeError

write() argument must be str, not {type(s).__name__}

Error message

write() argument must be str, not {type(s).__name__}

What it means

Debug-mode guard in BaseSelectorEventLoop.sock_recvfrom_into(): before performing an asynchronous datagram receive into a provided buffer, the loop checks that the socket is non-blocking when debug mode is enabled. Blocking sockets would hang the loop inside recvfrom_into, so the guard converts that into an immediate ValueError during development.

Source

Thrown at Lib/_apple_support.py:24

    # Redirect stdout and stderr to the Apple system log. This method is
    # invoked by init_apple_streams() (initconfig.c) if config->use_system_logger
    # is enabled.
    sys.stdout = SystemLog(log_write, stdout_level, errors=sys.stderr.errors)
    sys.stderr = SystemLog(log_write, stderr_level, errors=sys.stderr.errors)


class SystemLog(io.TextIOWrapper):
    def __init__(self, log_write, level, **kwargs):
        kwargs.setdefault("encoding", "UTF-8")
        kwargs.setdefault("line_buffering", True)
        super().__init__(LogStream(log_write, level), **kwargs)

    def __repr__(self):
        return f"<SystemLog (level {self.buffer.level})>"

    def write(self, s):
        if not isinstance(s, str):
            raise TypeError(
                f"write() argument must be str, not {type(s).__name__}")

        # In case `s` is a str subclass that writes itself to stdout or stderr
        # when we call its methods, convert it to an actual str.
        s = str.__str__(s)

        # We want to emit one log message per line, so split
        # the string before sending it to the superclass.
        for line in s.splitlines(keepends=True):
            super().write(line)

        return len(s)


class LogStream(io.RawIOBase):
    def __init__(self, log_write, level):
        self.log_write = log_write
        self.level = level

View on GitHub (pinned to bc6749cc3b)

Solutions

  1. Set sock.setblocking(False) before the first await on sock_recvfrom_into
  2. Migrate socket construction to an async-aware factory
  3. Where timeouts are genuinely wanted, implement them with asyncio.wait_for around the await instead of socket timeouts

Example fix

# before
sock.settimeout(2.0)
n, addr = await loop.sock_recvfrom_into(sock, buf)  # debug: ValueError
# after
sock.setblocking(False)
n, addr = await asyncio.wait_for(loop.sock_recvfrom_into(sock, buf), 2.0)
Defensive patterns

Strategy: validation

Validate before calling

def nb(sock):
    if sock.gettimeout() != 0:
        sock.setblocking(False)
    return sock

n, addr = await asyncio.wait_for(loop.sock_recvfrom_into(nb(sock), buf), 2.0)

Try / catch

try:
    await loop.sock_recvfrom_into(sock, buf, n)
except ValueError as e:
    if 'non-blocking' in str(e):
        sock.setblocking(False)
        return await loop.sock_recvfrom_into(sock, buf, n)
    raise

Prevention

When it happens

Trigger: await loop.sock_recvfrom_into(sock, buf, nbytes) with debug on and sock.gettimeout() != 0, covering both fully blocking sockets (timeout None) and those with explicit timeouts left from sync usage.

Common situations: Converting sync UDP proxy code that used recvfrom_into with settimeout for reliability; benchmark tools reusing capture sockets; debug-mode-enabled test matrices surfacing previously silent blocking state.

Related errors


AI-assisted analysis of python/cpython@bc6749cc3b (2026-08-14). Data as JSON: /api/errors/71cf235af2a6dc09. Report an issue: GitHub.