Write the buffer to the output file.
(self)
| 2057 | self.on_broken_pipe() |
| 2058 | |
| 2059 | def _write_buffer(self) -> None: |
| 2060 | """Write the buffer to the output file.""" |
| 2061 | |
| 2062 | with self._lock: |
| 2063 | if self.record and not self._buffer_index: |
| 2064 | with self._record_buffer_lock: |
| 2065 | self._record_buffer.extend(self._buffer[:]) |
| 2066 | |
| 2067 | if self._buffer_index == 0: |
| 2068 | if self.is_jupyter: # pragma: no cover |
| 2069 | from .jupyter import display |
| 2070 | |
| 2071 | display(self._buffer, self._render_buffer(self._buffer[:])) |
| 2072 | del self._buffer[:] |
| 2073 | else: |
| 2074 | if WINDOWS: |
| 2075 | use_legacy_windows_render = False |
| 2076 | if self.legacy_windows: |
| 2077 | fileno = get_fileno(self.file) |
| 2078 | if fileno is not None: |
| 2079 | use_legacy_windows_render = ( |
| 2080 | fileno in _STD_STREAMS_OUTPUT |
| 2081 | ) |
| 2082 | |
| 2083 | if use_legacy_windows_render: |
| 2084 | from rich._win32_console import LegacyWindowsTerm |
| 2085 | from rich._windows_renderer import legacy_windows_render |
| 2086 | |
| 2087 | buffer = self._buffer[:] |
| 2088 | if self.no_color and self._color_system: |
| 2089 | buffer = list(Segment.remove_color(buffer)) |
| 2090 | |
| 2091 | legacy_windows_render(buffer, LegacyWindowsTerm(self.file)) |
| 2092 | else: |
| 2093 | # Either a non-std stream on legacy Windows, or modern Windows. |
| 2094 | text = self._render_buffer(self._buffer[:]) |
| 2095 | # https://bugs.python.org/issue37871 |
| 2096 | # https://github.com/python/cpython/issues/82052 |
| 2097 | # We need to avoid writing more than 32Kb in a single write, due to the above bug |
| 2098 | write = self.file.write |
| 2099 | # Worse case scenario, every character is 4 bytes of utf-8 |
| 2100 | MAX_WRITE = 32 * 1024 // 4 |
| 2101 | try: |
| 2102 | if len(text) <= MAX_WRITE: |
| 2103 | write(text) |
| 2104 | else: |
| 2105 | batch: List[str] = [] |
| 2106 | batch_append = batch.append |
| 2107 | size = 0 |
| 2108 | for line in text.splitlines(True): |
| 2109 | if size + len(line) > MAX_WRITE and batch: |
| 2110 | write("".join(batch)) |
| 2111 | batch.clear() |
| 2112 | size = 0 |
| 2113 | batch_append(line) |
| 2114 | size += len(line) |
| 2115 | if batch: |
| 2116 | write("".join(batch)) |
no test coverage detected