Python typing 模块，IO 实例源码

我们从Python开源项目中，提取了以下46个代码示例，用于说明如何使用typing.IO。

项目：charm-glusterfs 作者：openstack | 项目源码 | 文件源码

def parse(f: IO[Any]) -> Result:
    """
    Parse a shellscript and return a ShellScript
    :param f: TextIOBase handle to the shellscript file
    :return: Result with Ok or Err
    """
    comments = []
    commands = []
    interpreter = ""

    buf = f.readlines()

    for line in buf:
        trimmed = line.strip()
        if trimmed.startswith("#!"):
            interpreter = trimmed
        elif trimmed.startswith("#"):
            comments.append(str(trimmed))
        else:
            # Skip blank lines
            if trimmed:
                commands.append(str(trimmed))
    return Ok(ShellScript(interpreter=interpreter,
                          comments=comments,
                          commands=commands))

项目：charm-glusterfs 作者：openstack | 项目源码 | 文件源码

def parse_entries(self, file: IO[Any]) -> Result:
        """
        Parse fstab entries
        :param file: TextIOWrapper file handle to the fstab
        :return: Result with Ok or Err
        """
        entries = []
        contents = file.readlines()

        for line in contents:
            if line.startswith("#"):
                continue
            parts = line.split()
            if len(parts) != 6:
                continue
            fsck_order = int(parts[5])
            entries.append(FsEntry(
                fs_spec=parts[0],
                mountpoint=os.path.join(parts[1]),
                vfs_type=parts[2],
                mount_options=parts[3].split(","),
                dump=False if parts[4] == "0" else True,
                fsck_order=fsck_order))
        return Ok(entries)

项目：py-aspio 作者：hexhex | 项目源码 | 文件源码

def __init__(self, *, process: subprocess.Popen, encoding: str, tmp_input: FilesystemIPC) -> None:
        self.process = process
        self.stdout_encoding = encoding
        self.iterating = False
        #
        # We need to capture stderr in a background thread to avoid deadlocks.
        # (The problem would occur when dlvhex2 is blocked because the OS buffers on the stderr pipe are full... so we have to constantly read from *both* stdout and stderr)
        self.stderr_capture_thread = StreamCaptureThread(self.process.stderr)
        self.stderr_capture_thread.start()
        #
        # Set up finalization. Using weakref.finalize seems to work more robustly than using __del__.
        # (One problem that occurred with __del__: It seemed like python was calling __del__ for self.process and its IO streams first,
        # which resulted in ResourceWarnings even though we were closing the streams properly in our __del__ function.)
        self._finalize = weakref.finalize(self, DlvhexLineReader.__close, process, self.stderr_capture_thread, encoding, tmp_input)  # type: ignore
        # Make sure the subprocess will be terminated if it's still running when the python process exits
        self._finalize.atexit = True

项目：pysport 作者：sportorg | 项目源码 | 文件源码

def parse(self, file):
        if not isinstance(file, str) and not isinstance(file, IO):
            raise TypeError("file is not str or IO")
        if isinstance(file, str):
            try:
                enc = 'windows-1251'
                with open(file, encoding=enc) as f:
                    content = f.readlines()
            except FileNotFoundError:
                raise FileNotFoundError("Not found " + file)
        else:
            content = file.readlines()
        self._data = [x.strip() for x in content]
        self.clear()

        return self

项目：binaryalert 作者：airbnb | 项目源码 | 文件源码

def _read_in_chunks(file_object: IO[bytes], chunk_size: int = 2*MB) -> Generator[bytes, None, None]:
    """Read a file in fixed-size chunks (to minimize memory usage for large files).

    Args:
        file_object: An opened file-like object supporting read().
        chunk_size: Max size (in bytes) of each file chunk.

    Yields:
        File chunks, each of size at most chunk_size.
    """
    while True:
        chunk = file_object.read(chunk_size)
        if chunk:
            yield chunk
        else:
            return  # End of file.

项目：yatta_reader 作者：sound88 | 项目源码 | 文件源码

def __init__(
        self,
        _=None,  # type: Optional[Union[AnyStr, typing.Mapping, typing.Sequence, typing.IO]]
    ):
        self._meta = None
        if _ is not None:
            if isinstance(_, HTTPResponse):
                meta.get(self).url = _.url
            _ = deserialize(_)
            for k, v in _.items():
                try:
                    self[k] = v
                except KeyError as e:
                    if e.args and len(e.args) == 1:
                        e.args = (
                            r'%s.%s: %s' % (type(self).__name__, e.args[0], json.dumps(_)),
                        )
                    raise e