newline=None, closefd=True, opener=None):

r"""Open file and return a stream. Raise OSError upon failure.

if not isinstance(file, int):

file = os.fspath(file)

if not isinstance(file, (str, bytes, int)):

raise TypeError("invalid file: %r" % file)

if not isinstance(mode, str):

raise TypeError("invalid mode: %r" % mode)

if not isinstance(buffering, int):

raise TypeError("invalid buffering: %r" % buffering)

if encoding is not None and not isinstance(encoding, str):

raise TypeError("invalid encoding: %r" % encoding)

if errors is not None and not isinstance(errors, str):

raise TypeError("invalid errors: %r" % errors)

modes = set(mode)

if modes - set("axrwb+tU") or len(mode) > len(modes):

raise ValueError("invalid mode: %r" % mode)

creating = "x" in modes

reading = "r" in modes

writing = "w" in modes

appending = "a" in modes

updating = "+" in modes

text = "t" in modes

binary = "b" in modes

if "U" in modes:

if creating or writing or appending or updating:

raise ValueError("mode U cannot be combined with 'x', 'w', 'a', or '+'")

import warnings

warnings.warn("'U' mode is deprecated",

DeprecationWarning, 2)

reading = True

if text and binary:

raise ValueError("can't have text and binary mode at once")

if creating + reading + writing + appending > 1:

raise ValueError("can't have read/write/append mode at once")

if not (creating or reading or writing or appending):

raise ValueError("must have exactly one of read/write/append mode")

if binary and encoding is not None:

raise ValueError("binary mode doesn't take an encoding argument")

if binary and errors is not None:

raise ValueError("binary mode doesn't take an errors argument")

if binary and newline is not None:

raise ValueError("binary mode doesn't take a newline argument")

raw = FileIO(file,

(creating and "x" or "") +

(reading and "r" or "") +

(writing and "w" or "") +

(appending and "a" or "") +

(updating and "+" or ""),

closefd, opener=opener)

result = raw

try:

line_buffering = False

if buffering == 1 or buffering < 0 and raw.isatty():

buffering = -1

line_buffering = True

if buffering < 0:

buffering = DEFAULT_BUFFER_SIZE

try:

bs = os.fstat(raw.fileno()).st_blksize

except (OSError, AttributeError):

pass

else:

if bs > 1:

buffering = bs

if buffering < 0:

raise ValueError("invalid buffering size")

if buffering == 0:

if binary:

return result

raise ValueError("can't have unbuffered text I/O")

if updating:

buffer = BufferedRandom(raw, buffering)

elif creating or writing or appending:

buffer = BufferedWriter(raw, buffering)

elif reading:

buffer = BufferedReader(raw, buffering)

else:

raise ValueError("unknown mode: %r" % mode)

result = buffer

if binary:

return result

text = TextIOWrapper(buffer, encoding, errors, newline, line_buffering)

result = text

text.mode = mode

return result

except:

result.close()

"""Helper for builtins.open.__doc__

def __get__(self, obj, typ):

return (

"open(file, mode='r', buffering=-1, encoding=None, "

"errors=None, newline=None, closefd=True)\n\n" +

"""Wrapper for builtins.open

__doc__ = DocDescriptor()

def __new__(cls, *args, **kwargs):

"""The abstract base class for all I/O classes, acting on streams of

### Internal ###

def _unsupported(self, name):

"""Internal: raise an OSError exception for unsupported operations."""

raise UnsupportedOperation("%s.%s() not supported" %

(self.__class__.__name__, name))

### Positioning ###

def seek(self, pos, whence=0):

"""Change stream position.

self._unsupported("seek")

def tell(self):

"""Return an int indicating the current stream position."""

return self.seek(0, 1)

def truncate(self, pos=None):

"""Truncate file to size bytes.

self._unsupported("truncate")

### Flush and close ###

def flush(self):

"""Flush write buffers, if applicable.

self._checkClosed()

# XXX Should this return the number of bytes written???

__closed = False

def close(self):

"""Flush and close the IO object.

if not self.__closed:

try:

self.flush()

finally:

self.__closed = True

def __del__(self):

"""Destructor. Calls close()."""

# The try/except block is in case this is called at program

# exit time, when it's possible that globals have already been

# deleted, and then the close() call might fail. Since

# there's nothing we can do about such failures and they annoy

# the end users, we suppress the traceback.

try:

self.close()

except:

pass

### Inquiries ###

def seekable(self):

"""Return a bool indicating whether object supports random access.

return False

def _checkSeekable(self, msg=None):

"""Internal: raise UnsupportedOperation if file is not seekable

if not self.seekable():

raise UnsupportedOperation("File or stream is not seekable."

if msg is None else msg)

def readable(self):

"""Return a bool indicating whether object was opened for reading.

return False

def _checkReadable(self, msg=None):

"""Internal: raise UnsupportedOperation if file is not readable

if not self.readable():

raise UnsupportedOperation("File or stream is not readable."

if msg is None else msg)

def writable(self):

"""Return a bool indicating whether object was opened for writing.

return False

def _checkWritable(self, msg=None):

"""Internal: raise UnsupportedOperation if file is not writable

if not self.writable():

raise UnsupportedOperation("File or stream is not writable."

if msg is None else msg)

@property

def closed(self):

"""closed: bool. True iff the file has been closed.

return self.__closed

def _checkClosed(self, msg=None):

"""Internal: raise a ValueError if file is closed

if self.closed:

raise ValueError("I/O operation on closed file."

if msg is None else msg)

### Context manager ###

def __enter__(self): # That's a forward reference

"""Context management protocol. Returns self (an instance of IOBase)."""

self._checkClosed()

return self

def __exit__(self, *args):

"""Context management protocol. Calls close()"""

self.close()

### Lower-level APIs ###

# XXX Should these be present even if unimplemented?

def fileno(self):

"""Returns underlying file descriptor (an int) if one exists.

self._unsupported("fileno")

def isatty(self):

"""Return a bool indicating whether this is an 'interactive' stream.

self._checkClosed()

return False

### Readline[s] and writelines ###

def readline(self, size=-1):

r"""Read and return a line of bytes from the stream.

# For backwards compatibility, a (slowish) readline().

if hasattr(self, "peek"):

def nreadahead():

readahead = self.peek(1)

if not readahead:

return 1

n = (readahead.find(b"\n") + 1) or len(readahead)

if size >= 0:

n = min(n, size)

return n

else:

def nreadahead():

return 1

if size is None:

size = -1

elif not isinstance(size, int):

raise TypeError("size must be an integer")

res = bytearray()

while size < 0 or len(res) < size:

b = self.read(nreadahead())

if not b:

break

res += b

if res.endswith(b"\n"):

break

return bytes(res)

def __iter__(self):

self._checkClosed()

return self

def __next__(self):

line = self.readline()

if not line:

raise StopIteration

return line

def readlines(self, hint=None):

"""Return a list of lines from the stream.

if hint is None or hint <= 0:

return list(self)

n = 0

lines = []

for line in self:

lines.append(line)

n += len(line)

if n >= hint:

break

return lines

def writelines(self, lines):

self._checkClosed()

for line in lines:

"""Base class for raw binary I/O."""

# The read() method is implemented by calling readinto(); derived

# classes that want to support read() only need to implement

# readinto() as a primitive operation. In general, readinto() can be

# more efficient than read().

# (It would be tempting to also provide an implementation of

# readinto() in terms of read(), in case the latter is a more suitable

# primitive operation, but that would lead to nasty recursion in case

# a subclass doesn't implement either.)

def read(self, size=-1):

"""Read and return up to size bytes, where size is an int.

if size is None:

size = -1

if size < 0:

return self.readall()

b = bytearray(size.__index__())

n = self.readinto(b)

if n is None:

return None

del b[n:]

return bytes(b)

def readall(self):

"""Read until EOF, using multiple read() call."""

res = bytearray()

while True:

data = self.read(DEFAULT_BUFFER_SIZE)

if not data:

break

res += data

if res:

return bytes(res)

else:

# b'' or None

return data

def readinto(self, b):

"""Read bytes into a pre-allocated bytes-like object b.

self._unsupported("readinto")

def write(self, b):

"""Write the given buffer to the IO stream.

"""Base class for buffered IO objects.

def read(self, size=-1):

"""Read and return up to size bytes, where size is an int.

self._unsupported("read")

def read1(self, size=-1):

"""Read up to size bytes with at most one read() system call,

self._unsupported("read1")

def readinto(self, b):

"""Read bytes into a pre-allocated bytes-like object b.

return self._readinto(b, read1=False)

def readinto1(self, b):

"""Read bytes into buffer *b*, using at most one system call

return self._readinto(b, read1=True)

def _readinto(self, b, read1):

if not isinstance(b, memoryview):

b = memoryview(b)

b = b.cast('B')

if read1:

data = self.read1(len(b))

else:

data = self.read(len(b))

n = len(data)

b[:n] = data

return n

def write(self, b):

"""Write the given bytes buffer to the IO stream.

self._unsupported("write")

def detach(self):

"""

"""A mixin implementation of BufferedIOBase with an underlying raw stream.

def __init__(self, raw):

self._raw = raw

### Positioning ###

def seek(self, pos, whence=0):

new_position = self.raw.seek(pos, whence)

if new_position < 0:

raise OSError("seek() returned an invalid position")

return new_position

def tell(self):

pos = self.raw.tell()

if pos < 0:

raise OSError("tell() returned an invalid position")

return pos

def truncate(self, pos=None):

# Flush the stream. We're mixing buffered I/O with lower-level I/O,

# and a flush may be necessary to synch both views of the current

# file state.

self.flush()

if pos is None:

pos = self.tell()

# XXX: Should seek() be used, instead of passing the position

# XXX directly to truncate?

return self.raw.truncate(pos)

### Flush and close ###

def flush(self):

if self.closed:

raise ValueError("flush on closed file")

self.raw.flush()

def close(self):

if self.raw is not None and not self.closed:

try:

# may raise BlockingIOError or BrokenPipeError etc

self.flush()

finally:

self.raw.close()

def detach(self):

if self.raw is None:

raise ValueError("raw stream already detached")

self.flush()

raw = self._raw

self._raw = None

return raw

### Inquiries ###

def seekable(self):

return self.raw.seekable()

@property

def raw(self):

return self._raw

@property

def closed(self):

return self.raw.closed

@property

def name(self):

return self.raw.name

@property

def mode(self):

return self.raw.mode

def __getstate__(self):

raise TypeError("can not serialize a '{0}' object"

.format(self.__class__.__name__))

def __repr__(self):

modname = self.__class__.__module__

clsname = self.__class__.__qualname__

try:

name = self.name

except Exception:

return "<{}.{}>".format(modname, clsname)

else:

return "<{}.{} name={!r}>".format(modname, clsname, name)

### Lower-level APIs ###

def fileno(self):

return self.raw.fileno()

def isatty(self):

"""Buffered I/O implementation using an in-memory bytes buffer."""

def __init__(self, initial_bytes=None):

buf = bytearray()

if initial_bytes is not None:

buf += initial_bytes

self._buffer = buf

self._pos = 0

def __getstate__(self):

if self.closed:

raise ValueError("__getstate__ on closed file")

return self.__dict__.copy()

def getvalue(self):

"""Return the bytes value (contents) of the buffer

if self.closed:

raise ValueError("getvalue on closed file")

return bytes(self._buffer)

def getbuffer(self):

"""Return a readable and writable view of the buffer.

if self.closed:

raise ValueError("getbuffer on closed file")

return memoryview(self._buffer)

def close(self):

self._buffer.clear()

super().close()

def read(self, size=-1):

if self.closed:

raise ValueError("read from closed file")

if size is None:

size = -1

if size < 0:

size = len(self._buffer)

if len(self._buffer) <= self._pos:

return b""

newpos = min(len(self._buffer), self._pos + size)

b = self._buffer[self._pos : newpos]

self._pos = newpos

return bytes(b)

def read1(self, size=-1):

"""This is the same as read.

return self.read(size)

def write(self, b):

if self.closed:

raise ValueError("write to closed file")

if isinstance(b, str):

raise TypeError("can't write str to binary stream")

with memoryview(b) as view:

n = view.nbytes # Size of any bytes-like object

if n == 0:

return 0

pos = self._pos

if pos > len(self._buffer):

# Inserts null bytes between the current end of the file

# and the new write position.

padding = b'\x00' * (pos - len(self._buffer))

self._buffer += padding

self._buffer[pos:pos + n] = b

self._pos += n

return n

def seek(self, pos, whence=0):

if self.closed:

raise ValueError("seek on closed file")

try:

pos.__index__

except AttributeError as err:

raise TypeError("an integer is required") from err

if whence == 0:

if pos < 0:

raise ValueError("negative seek position %r" % (pos,))

self._pos = pos

elif whence == 1:

self._pos = max(0, self._pos + pos)

elif whence == 2:

self._pos = max(0, len(self._buffer) + pos)

else:

raise ValueError("unsupported whence value")

return self._pos

def tell(self):

if self.closed:

raise ValueError("tell on closed file")

return self._pos

def truncate(self, pos=None):

if self.closed:

raise ValueError("truncate on closed file")

if pos is None:

pos = self._pos

else:

try:

pos.__index__

except AttributeError as err:

raise TypeError("an integer is required") from err

if pos < 0:

raise ValueError("negative truncate position %r" % (pos,))

del self._buffer[pos:]

return pos

def readable(self):

if self.closed:

raise ValueError("I/O operation on closed file.")

return True

def writable(self):

if self.closed:

raise ValueError("I/O operation on closed file.")

return True

def seekable(self):

if self.closed:

raise ValueError("I/O operation on closed file.")

"""BufferedReader(raw[, buffer_size])

def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):

"""Create a new buffered reader using the given readable raw IO object.

if not raw.readable():

raise OSError('"raw" argument must be readable.')

_BufferedIOMixin.__init__(self, raw)

if buffer_size <= 0:

raise ValueError("invalid buffer size")

self.buffer_size = buffer_size

self._reset_read_buf()

self._read_lock = Lock()

def readable(self):

return self.raw.readable()

def _reset_read_buf(self):

self._read_buf = b""

self._read_pos = 0

def read(self, size=None):

"""Read size bytes.

if size is not None and size < -1:

raise ValueError("invalid number of bytes to read")

with self._read_lock: