Read & decode bytes from a local stdin stream. :param input_: Actual stream object to read from. Maps to ``in_stream`` in `run`, so will often be ``sys.stdin``, but might be any stream-like object. :returns: A Unicode string,
(self, input_: IO)
| 810 | ) |
| 811 | |
| 812 | def read_our_stdin(self, input_: IO) -> Optional[str]: |
| 813 | """ |
| 814 | Read & decode bytes from a local stdin stream. |
| 815 | |
| 816 | :param input_: |
| 817 | Actual stream object to read from. Maps to ``in_stream`` in `run`, |
| 818 | so will often be ``sys.stdin``, but might be any stream-like |
| 819 | object. |
| 820 | |
| 821 | :returns: |
| 822 | A Unicode string, the result of decoding the read bytes (this might |
| 823 | be the empty string if the pipe has closed/reached EOF); or |
| 824 | ``None`` if stdin wasn't ready for reading yet. |
| 825 | |
| 826 | .. versionadded:: 1.0 |
| 827 | """ |
| 828 | # TODO: consider moving the character_buffered contextmanager call in |
| 829 | # here? Downside is it would be flipping those switches for every byte |
| 830 | # read instead of once per session, which could be costly (?). |
| 831 | bytes_ = None |
| 832 | if ready_for_reading(input_): |
| 833 | try: |
| 834 | bytes_ = input_.read(bytes_to_read(input_)) |
| 835 | except OSError as e: |
| 836 | # Assume EBADF in this situation implies running under nohup or |
| 837 | # similar, where: |
| 838 | # - we cannot reliably detect a bad FD up front |
| 839 | # - trying to read it would explode |
| 840 | # - user almost surely doesn't care about stdin anyways |
| 841 | # and ignore it (but not other OSErrors!) |
| 842 | if e.errno != errno.EBADF: |
| 843 | raise |
| 844 | # Decode if it appears to be binary-type. (From real terminal |
| 845 | # streams, usually yes; from file-like objects, often no.) |
| 846 | if bytes_ and isinstance(bytes_, bytes): |
| 847 | # TODO: will decoding 1 byte at a time break multibyte |
| 848 | # character encodings? How to square interactivity with that? |
| 849 | bytes_ = self.decode(bytes_) |
| 850 | return bytes_ |
| 851 | |
| 852 | def handle_stdin( |
| 853 | self, |
no test coverage detected