#define PY_STDIOTEXTMODE "b"
PyAPI_FUNC(PyObject *) PyFile_FromFd(int, char *, char *, int, char *, char *,
- int);
+ char *, int);
PyAPI_FUNC(PyObject *) PyFile_GetLine(PyObject *, int);
PyAPI_FUNC(int) PyFile_WriteObject(PyObject *, PyObject *, int);
PyAPI_FUNC(int) PyFile_WriteString(const char *, PyObject *);
self.characters_written = characters_written
-def open(file, mode="r", buffering=None, encoding=None, newline=None,
- closefd=True):
+def open(file, mode="r", buffering=None, encoding=None, errors=None,
+ newline=None, closefd=True):
r"""Replacement for the built-in open function.
Args:
can be: 0 = unbuffered, 1 = line buffered,
larger = fully buffered.
encoding: optional string giving the text encoding.
+ errors: optional string giving the encoding error handling.
newline: optional newlines specifier; must be None, '', '\n', '\r'
or '\r\n'; all other values are illegal. It controls the
handling of line endings. It works as follows:
'U': universal newline mode (for backwards compatibility)
Constraints:
- - encoding must not be given when a binary mode is given
+ - encoding or errors must not be given when a binary mode is given
- buffering must not be zero when a text mode is given
Returns:
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("arwb+tU") or len(mode) > len(modes):
raise ValueError("invalid mode: %r" % mode)
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,
buffer.name = file
buffer.mode = mode
return buffer
- text = TextIOWrapper(buffer, encoding, newline)
+ text = TextIOWrapper(buffer, encoding, errors, newline)
text.name = file
text.mode = mode
return text
_CHUNK_SIZE = 128
- def __init__(self, buffer, encoding=None, newline=None):
+ def __init__(self, buffer, encoding=None, errors=None, newline=None):
if newline not in (None, "", "\n", "\r", "\r\n"):
raise ValueError("illegal newline value: %r" % (newline,))
if encoding is None:
if not isinstance(encoding, str):
raise ValueError("invalid encoding: %r" % encoding)
+ if errors is None:
+ errors = "strict"
+ else:
+ if not isinstance(errors, str):
+ raise ValueError("invalid errors: %r" % errors)
+
self.buffer = buffer
self._encoding = encoding
+ self._errors = errors
self._readuniversal = not newline
self._readtranslate = newline is None
self._readnl = newline
def encoding(self):
return self._encoding
+ @property
+ def errors(self):
+ return self._errors
+
# A word about _snapshot. This attribute is either None, or a
# tuple (decoder_state, readahead, pending) where decoder_state is
# the second (integer) item of the decoder state, readahead is the
if haslf and self._writetranslate and self._writenl != "\n":
s = s.replace("\n", self._writenl)
# XXX What if we were just reading?
- b = s.encode(self._encoding)
+ b = s.encode(self._encoding, self._errors)
self.buffer.write(b)
if haslf and self.isatty():
self.flush()
if make_decoder is None:
raise IOError("Can't find an incremental decoder for encoding %s" %
self._encoding)
- decoder = make_decoder() # XXX: errors
+ decoder = make_decoder(self._errors)
if self._readuniversal:
decoder = IncrementalNewlineDecoder(decoder, self._readtranslate)
self._decoder = decoder
# XXX This is really slow, but fully functional
- def __init__(self, initial_value="", encoding="utf-8", newline="\n"):
+ def __init__(self, initial_value="", encoding="utf-8",
+ errors="strict", newline="\n"):
super(StringIO, self).__init__(BytesIO(),
encoding=encoding,
+ errors=errors,
newline=newline)
if initial_value:
if not isinstance(initial_value, str):
def getvalue(self):
self.flush()
- return self.buffer.getvalue().decode(self._encoding)
+ return self.buffer.getvalue().decode(self._encoding, self._errors)
def tearDown(self):
test_support.unlink(test_support.TESTFN)
+ def testEncodingErrorsReading(self):
+ # (1) default
+ b = io.BytesIO(b"abc\n\xff\n")
+ t = io.TextIOWrapper(b, encoding="ascii")
+ self.assertRaises(UnicodeError, t.read)
+ # (2) explicit strict
+ b = io.BytesIO(b"abc\n\xff\n")
+ t = io.TextIOWrapper(b, encoding="ascii", errors="strict")
+ self.assertRaises(UnicodeError, t.read)
+ # (3) ignore
+ b = io.BytesIO(b"abc\n\xff\n")
+ t = io.TextIOWrapper(b, encoding="ascii", errors="ignore")
+ self.assertEquals(t.read(), "abc\n\n")
+ # (4) replace
+ b = io.BytesIO(b"abc\n\xff\n")
+ t = io.TextIOWrapper(b, encoding="ascii", errors="replace")
+ self.assertEquals(t.read(), "abc\n\ufffd\n")
+
+ def testEncodingErrorsWriting(self):
+ # (1) default
+ b = io.BytesIO()
+ t = io.TextIOWrapper(b, encoding="ascii")
+ self.assertRaises(UnicodeError, t.write, "\xff")
+ # (2) explicit strict
+ b = io.BytesIO()
+ t = io.TextIOWrapper(b, encoding="ascii", errors="strict")
+ self.assertRaises(UnicodeError, t.write, "\xff")
+ # (3) ignore
+ b = io.BytesIO()
+ t = io.TextIOWrapper(b, encoding="ascii", errors="ignore")
+ t.write("abc\xffdef\n")
+ t.flush()
+ self.assertEquals(b.getvalue(), b"abcdef\n")
+ # (4) replace
+ b = io.BytesIO()
+ t = io.TextIOWrapper(b, encoding="ascii", errors="replace")
+ t.write("abc\xffdef\n")
+ t.flush()
+ self.assertEquals(b.getvalue(), b"abc?def\n")
+
def testNewlinesInput(self):
testdata = b"AAA\nBBB\nCCC\rDDD\rEEE\r\nFFF\r\nGGG"
normalized = testdata.replace(b"\r\n", b"\n").replace(b"\r", b"\n")
return NULL;
}
cr.real = PyFloat_AsDouble(tmp);
+ cr.imag = 0.0; /* Shut up compiler warning */
Py_DECREF(tmp);
}
if (i == NULL) {
PyObject *
PyFile_FromFd(int fd, char *name, char *mode, int buffering, char *encoding,
- char *newline, int closefd)
+ char *errors, char *newline, int closefd)
{
PyObject *io, *stream, *nameobj = NULL;
io = PyImport_ImportModule("io");
if (io == NULL)
return NULL;
- stream = PyObject_CallMethod(io, "open", "isissi", fd, mode,
- buffering, encoding, newline, closefd);
+ stream = PyObject_CallMethod(io, "open", "isisssi", fd, mode,
+ buffering, encoding, errors,
+ newline, closefd);
Py_DECREF(io);
if (stream == NULL)
return NULL;
(char*)PyUnicode_GetDefaultEncoding();
}
fob = PyFile_FromFd(fd, pathname, fdp->mode, -1,
- (char*)encoding, NULL, 1);
+ (char*)encoding, NULL, NULL, 1);
if (fob == NULL) {
close(fd);
PyMem_FREE(found_encoding);
#endif
}
else {
- if (!(std = PyFile_FromFd(fd, "<stdin>", "r", -1, NULL,
+ if (!(std = PyFile_FromFd(fd, "<stdin>", "r", -1, NULL, NULL,
"\n", 0))) {
goto error;
}
#endif
}
else {
- if (!(std = PyFile_FromFd(fd, "<stdout>", "w", -1, NULL,
+ if (!(std = PyFile_FromFd(fd, "<stdout>", "w", -1, NULL, NULL,
"\n", 0))) {
goto error;
}
#endif
}
else {
- if (!(std = PyFile_FromFd(fd, "<stderr>", "w", -1, NULL,
+ if (!(std = PyFile_FromFd(fd, "<stderr>", "w", -1, NULL, NULL,
"\n", 0))) {
goto error;
}
projects / python / commitdiff