[Python-checkins] r45823 - sandbox/trunk/sio/io.py
guido.van.rossum
python-checkins at python.org
Sun Apr 30 18:01:39 CEST 2006
Author: guido.van.rossum
Date: Sun Apr 30 18:01:38 2006
New Revision: 45823
Added:
sandbox/trunk/sio/io.py (contents, props changed)
Log:
Check in a very early draft for a Py3K I/O library.
Don't look yet.
Added: sandbox/trunk/sio/io.py
==============================================================================
--- (empty file)
+++ sandbox/trunk/sio/io.py Sun Apr 30 18:01:38 2006
@@ -0,0 +1,245 @@
+"""New I/O library for Python 3.0.
+
+This is inspired by sandbox/sio/sio.py but uses bytes for low-level
+I/O and unicode strings for high-level I/O.
+
+"""
+
+# XXX What about thread-safety?
+
+__all__ = ["open"]
+
+import os
+
+DEFAULT_BUFSIZE = 8*1024 # International standard buffer size
+
+class IBinaryFile:
+
+ """Abstract base class documenting the API for a binary file.
+
+ All operations may raise IOError, OSError or some other low-level
+ I/O-related exception (e.g. socket.error).
+
+ """
+
+ def readable(self):
+ """Returns True if the file supports the read() operation."""
+
+ def writable(self):
+ """Returns True if the file supports the write() operation."""
+
+ def seekable(self):
+ """Returns True if the file supports seek() and tell()."""
+
+ def truncatable(self):
+ """Returns True if the file supports truncate()."""
+
+ def read(self, n):
+ """Reads up to n bytes from a readable file.
+
+ Returns a bytes object whose length is <= n.
+
+ The returned object is empty only if EOF is reached.
+
+ A short read is nothing unusal and doesn't mean that the next
+ read will report EOF.
+
+ """
+
+ def write(self, b):
+ """Writes b, which must be a bytes object.
+
+ Not all bytes may be written. At least one byte is written
+ unless b is empty.
+
+ The return value is the number of bytes written; it is only
+ zero when b is empty. (When no bytes are written, an
+ exception is raised.)
+
+ """
+
+ def tell(self):
+ """Returns the current byte offset from the start of the file."""
+
+ def seek(self, offset, whence=0):
+ """Positions the file at the given byte offset.
+
+ If whence is 0, the offset is measured from the start of the
+ file. If it is 1, the offset is measured from the current
+ position. If it is 2, the offset is measured from the end of
+ the file.
+
+ A negative offset may be specified if whence is 1 or 2 but the
+ effect is undefined if the resulting position is negative or
+ beyond EOF.
+
+ Returns the new byte offset.
+
+ """
+
+ def truncate(self, offset):
+ """Truncates the file at the given byte offset.
+
+ Sets the file position to the new EOF position.
+
+ The effect is undefined if offset is negative or beyond EOF.
+
+ Returns the new file size, in bytes.
+
+ """
+
+
+class OSBinaryFile:
+
+ """A binary file using I/O on Unix file descriptors."""
+
+ def __init__(self, fd):
+ self._fd = fd
+
+ def readable(self):
+ return True # May be a lie -- but how to tell the mode?
+
+ def writable(self):
+ return True # May be a lie -- but how to tell the mode?
+
+ def seekable(self):
+ return True # May be a lie -- but how to tell the mode?
+
+ def truncatable(self):
+ return True # May be a lie -- but how to tell the mode?
+
+ def read(self, n):
+ b = os.read(self._fd, n)
+ if not isinstance(b, bytes):
+ b = bytes(b)
+ return b
+
+ def write(self, b):
+ assert isinstance(b, bytes)
+ return os.write(self._fd, b)
+
+ def seek(self, pos, whence=0):
+ return os.lseek(self._fd, pos, whence)
+
+ def tell(self):
+ return os.lseek(self._fd, 0, 1)
+
+ def truncate(self, offset):
+ os.ftruncate(self._fd, offset)
+ return os.lseek(self._fd, 0, 2)
+
+ # XXX ioctl, fcntl? What else?
+
+
+class BufferingBinaryFileWrapper:
+
+ """A buffering wrapper for a binary file.
+
+ This provides exactly the same API but uses internal buffering to
+ speed up small reads and writes. An additional flush() method
+ forces data out to the underlying binary file or throws away
+ unused read-ahead data.
+
+ WHen this is used for reading *and* writing, it assumes the
+ underlying file is seekable. For a socket, you need to open two
+ separate buffering wrappers: one for reading, and a separate one
+ for writing.
+
+ """
+
+ def __init__(self, file, bufsize=DEFAULT_BUFSIZE):
+ bufsize = bufsize.__index__()
+ if bufsize <= 0:
+ raise ValueError("bufsize must be > 0, not %d" % bufsize)
+ self._file = file
+ self._bufsize = bufsize
+ self._buffer = bytes()
+ self._bufptr = 0
+ self._writing = False
+ # Invariants:
+ # 0 <= self._bufptr <= len(self._buffer) <= self._bufsize
+ # Meaning of self._bufptr:
+ # if self._writing:
+ # self._buffer[ : self._bufptr] = data to be flushed
+ # else:
+ # self._buffer[self._bufptr : ] = data to be read
+
+ def flush(self):
+ if self._writing:
+ start = 0
+ while start < self._bufptr:
+ start += self._file.write(self._buffer[start : self._bufptr])
+ self._bufptr = 0
+ elif self._bufptr < len(self._buffer):
+ self._file.seek(self._bufptr - len(self._buffer), 1)
+ self._buffer = bytes()
+ self._bufptr = 0
+
+ def read(self, n):
+ n = n.__index__()
+ if not self._file.readable():
+ raise IOError("file is not open for reading")
+ if self._writing:
+ self.flush()
+ self._writing = False
+ result = None
+ while n > 0:
+ if self._bufptr < len(self._buffer):
+ data = self._buffer[self._bufptr : self._bufptr + n]
+ self._bufptr += len(data)
+ n -= len(data)
+ if result is None:
+ result = data
+ else:
+ result += data
+ else:
+ self._buffer = self._file.read(max(n, self._bufsize))
+ self._bufptr = 0
+ if len(self._buffer) == 0:
+ break
+ if result is None:
+ result = bytes()
+ return result
+
+ def write(self, b):
+ if not isinstance(b, bytes):
+ raise TypeError("write() requires bytes, not %r" % type(b))
+ if not self._file.writable():
+ raise IOError("file is not open for writing")
+ if not self._writing:
+ self.flush()
+ self._writing = True
+ if ((self._bufptr == 0 and len(b) >= self._bufsize)
+ or
+ (self._bufptr + len(b) >= 2*self._bufsize)):
+ self.flush()
+ start = 0
+ while start < len(b):
+ if start == 0:
+ start += self._file.write(b)
+ else:
+ start += self._file.write(b[start : ])
+ return len(b)
+ if self._bufptr == 0 and not self._buffer:
+ self._buffer = bytes(self._bufsize)
+ if self._bufptr + len(b) < len(self._buffer):
+ self._buffer[self._bufptr : self._bufptr + len(b)] = b
+ elif self._bufptr + len(b) == len(self._buffer):
+ self._buffer[self._bufptr : ] = b
+ self.flush()
+ else:
+ n = len(self._buffer) - self._bufptr
+ self._buffer[self._bufptr : ] = b[ : n]
+ self.flush()
+ self._bufptr[ : len(b) - n] = b[n : ]
+ self._bufptr = len(b) - n
+
+
+def main():
+ import sys
+ fd = sys.stdout.fileno()
+ f = OSBinaryFile(fd)
+ f.write(bytes(u"hello world\n", "ascii"))
+
+if __name__ == "__main__":
+ main()
More information about the Python-checkins
mailing list