{

friend class ::Heap;

private:

IOBase() : IOBase(s_io_base->type()) {}

protected:

IOBase(const PyType *type) : PyBaseObject(type->underlying_type()) {}

private:

PyResult<bool> is_closed() const

{

auto result = lookup_attribute(PyString::create("__IOBase_closed").unwrap());

if (std::get<0>(result).is_err()) return Err(std::get<0>(result).unwrap_err());

return Ok(std::get<1>(result) == LookupAttrResult::FOUND);

}

PyResult<std::monostate> check_closed() const

{

auto closed = lookup_attribute(PyString::create("closed").unwrap());

if (std::get<0>(closed).is_err()) return Err(std::get<0>(closed).unwrap_err());

if (std::get<1>(closed) == LookupAttrResult::FOUND) {

return truthy(std::get<0>(closed).unwrap(), VirtualMachine::the().interpreter())

.and_then([](bool closed) -> PyResult<std::monostate> {

if (!closed) return Ok(std::monostate{});

return Err(value_error("I/O operation on closed file."));

});

} else {

return Ok(std::monostate{});

}

}

public:

static constexpr std::string_view __doc__ =

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

"bytes. There is no public constructor.\n"

"\n"

"This class provides dummy implementations for many methods that\n"

"derived classes can override selectively; the default implementations\n"

"represent a file that cannot be read, written or seeked.\n"

"\n"

"Even though IOBase does not declare read, readinto, or write because\n"

"their signatures will vary, implementations and clients should\n"

"consider those methods part of the interface. Also, implementations\n"

"may raise UnsupportedOperation when operations they do not support are\n"

"called.\n"

"\n"

"The basic type used for binary data read from or written to a file is\n"

"bytes. Other bytes-like objects are accepted as method arguments too.\n"

"In some cases (such as readinto), a writable object is required. Text\n"

"I/O classes work with str data.\n"

"\n"

"Note that calling any method (except additional calls to close(),\n"

"which are ignored) on a closed stream should raise a ValueError.\n"

"\n"

"IOBase (and its subclasses) support the iterator protocol, meaning\n"

"that an IOBase object can be iterated over yielding the lines in a\n"

"stream.\n"

"\n"

"IOBase also supports the :keyword:`with` statement. In this example,\n"

"fp is closed after the suite of the with statement is complete:\n"

"\n"

"with open('spam.txt', 'r') as fp:\n"

" fp.write('Spam and eggs!')\n";

static PyResult<IOBase *> create(const PyType *type)

{

auto &heap = VirtualMachine::the().heap();

auto *result = heap.allocate<IOBase>(type);

if (!result) { return Err(memory_error(sizeof(IOBase))); }

return Ok(result);

}

static PyResult<PyObject *> __new__(const PyType *type, PyTuple *, PyDict *)

{

return IOBase::create(type);

}

PyResult<PyObject *> __iter__() const

{

if (auto closed = check_closed(); closed.is_err()) return Err(closed.unwrap_err());

return Ok(const_cast<IOBase *>(this));

}

PyResult<PyObject *> __next__()

{

return get_method(PyString::create("readline").unwrap()).and_then([](PyObject *readline) {

return readline->call(PyTuple::create().unwrap(), PyDict::create().unwrap())

.and_then([](PyObject *line) -> PyResult<PyObject *> {

auto m = line->as_mapping();

if (auto size = m.unwrap().len(); size.is_err()) {

return Err(size.unwrap_err());

} else {

if (size.unwrap() == 0) {

// since we don't handle the situation where __next__ returns

// Ok(nullptr), we make an extra allocation here for StopIteration,

// and have the same semantics as cpython

return Err(stop_iteration());

} else {

return Ok(line);

}

}

});

});

}

PyResult<PyObject *> close() const

{

auto closed = is_closed();

if (closed.is_err()) return Err(closed.unwrap_err());

if (closed.unwrap()) return Ok(py_none());

auto flushed =

get_method(PyString::create("flush").unwrap()).and_then([](PyObject *method) {

return method->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

if (flushed.is_err()) return flushed;

return Ok(py_none());

}

PyResult<PyObject *> fileno() const

{

return Err(unsupported_operation(PyTuple::create(String{ "fileno" }).unwrap(), nullptr));

}

PyResult<PyObject *> flush() const

{

auto closed = is_closed();

if (closed.is_err()) return Err(closed.unwrap_err());

if (closed.unwrap()) { return Err(value_error("I/O operation on closed file.")); }

return Ok(py_none());

}

PyResult<PyObject *> isatty() const

{

return check_closed().and_then([](auto) { return Ok(py_false()); });

}

PyResult<PyObject *> readline(int64_t limit) const

{

auto peek = lookup_attribute(PyString::create("peek").unwrap());

if (std::get<0>(peek).is_err()) return std::get<0>(peek);

const bool peakable = std::get<1>(peek) == LookupAttrResult::FOUND;

std::vector<std::byte> buffer;

while (limit < 0 || static_cast<int64_t>(buffer.size()) < limit) {

int64_t nreadahead = 1;

if (peakable) {

auto readahead = std::get<0>(peek).unwrap()->call(

PyTuple::create(Number{ 1 }).unwrap(), PyDict::create().unwrap());

if (readahead.is_err()) return readahead;

if (!as<PyBytes>(readahead.unwrap())) {

// FIXME: should be a OSError

return Err(value_error("peek() should have returned a bytes object, not '{}'",

readahead.unwrap()->type()->name()));

}

auto readahead_bytes = as<PyBytes>(readahead.unwrap());

if (!readahead_bytes->value().b.empty()) {

const auto &bytes = readahead_bytes->value().b;

const auto upper = limit == -1

? bytes.size()

: std::min(static_cast<int64_t>(bytes.size()), limit);

auto it = std::find(bytes.begin(), bytes.begin() + upper, std::byte{ '\n' });

nreadahead = std::distance(bytes.begin(), it);

}

}

auto b = get_method(PyString::create("read").unwrap())

.and_then([nreadahead](PyObject *read) {

return read->call(PyTuple::create(Number{ nreadahead }).unwrap(),

PyDict::create().unwrap());

});

if (b.is_err()) return b;

if (!as<PyBytes>(b.unwrap())) {

// FIXME: should be a OSError

return Err(value_error("read() should have returned a bytes object, not '{}'",

b.unwrap()->type()->name()));

}

auto *new_bytes = as<PyBytes>(b.unwrap());

if (new_bytes->value().b.size() == 0) { break; }

buffer.insert(buffer.end(), new_bytes->value().b.begin(), new_bytes->value().b.end());

if (static_cast<char>(buffer.back()) == '\n') { break; }

}

return PyBytes::create(Bytes{ std::move(buffer) });

}

PyResult<PyObject *> readlines(int64_t hint) const

{

auto result_ = PyList::create();

if (result_.is_err()) return result_;

auto *result = result_.unwrap();

auto it_ = iter();

if (it_.is_err()) return it_;

auto *it = it_.unwrap();

size_t length = 0;

while (true) {

auto line = it->next();

if (line.is_err()) {

if (line.unwrap_err()->type() == stop_iteration()->type()) {

break;

} else {

return line;

}

}

result->elements().push_back(line.unwrap());

if (hint > 0) {

if (auto m = result->as_mapping(); m.is_ok()) {

if (auto size = m.unwrap().len(); size.is_err()) {

return Err(size.unwrap_err());

} else {

length += size.unwrap();

if (static_cast<int64_t>(length) > hint) { break; }

}

} else {

return Err(m.unwrap_err());

}

}

}

return Ok(result);

}

PyResult<PyObject *> seek() const

{

// FIXME

return Err(unsupported_operation(PyTuple::create(String{ "seek" }).unwrap(), nullptr));

}

PyResult<PyObject *> seekable() const { return Ok(py_false()); }

PyResult<PyObject *> tell() const

{

return get_method(PyString::create("seek").unwrap()).and_then([](auto *seek) {

return seek->call(

PyTuple::create(Number{ 0 }, Number{ 1 }).unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> truncate() const

{

return Err(unsupported_operation(PyTuple::create(String{ "truncate" }).unwrap(), nullptr));

}

PyResult<PyObject *> writable() const { return Ok(py_false()); }

PyResult<PyObject *> writelines(PyObject *lines) const

{

if (auto closed = check_closed(); closed.is_err()) return Err(closed.unwrap_err());

auto iter_ = lines->iter();

if (iter_.is_err()) return iter_;

auto *iter = iter_.unwrap();

while (true) {

auto line = iter->next();

if (line.is_err()) {

if (line.unwrap_err()->type() == stop_iteration()->type()) {

break;

} else {

return line;

}

}

auto write = get_method(PyString::create("write").unwrap());

if (write.is_err()) return write;

auto res = write.unwrap()->call(

PyTuple::create(line.unwrap()).unwrap(), PyDict::create().unwrap());

if (res.is_err()) return res;

}

return Ok(py_none());

}

PyResult<PyObject *> check_seekable_() { return check_seekable(this); }

PyResult<PyObject *> check_readable_() { return check_readable(this); }

PyResult<PyObject *> check_writable_() { return check_writable(this); }

PyObject *dict() const { return m_attributes; }

PyType *static_type() const override { return s_io_base; }

PyResult<PyObject *> __enter__(PyTuple *, PyDict *)

{

return check_closed()

.and_then([this](auto) -> PyResult<PyObject *> { return Ok(this); })

.or_else([](auto *err) -> PyResult<PyObject *> { return Err(err); });

}

PyResult<PyObject *> __exit__(PyTuple *, PyDict *)

{

return get_method(PyString::create("close").unwrap()).and_then([](PyObject *close) {

return close->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

static PyType *register_type(PyModule *module)

{

if (!s_io_base) {

s_io_base =

klass<IOBase>(module, "_IOBase")

.def("close", &IOBase::close)

.property_readonly("closed",

[](IOBase *self) {

auto closed = self->is_closed();

if (closed.is_err()) { TODO(); }

return Ok(closed.unwrap() ? py_true() : py_false());

})

.property_readonly("__dict__", [](IOBase *self) { return Ok(self->dict()); })

.def("fileno", &IOBase::fileno)

.def("flush", &IOBase::flush)

.def("isatty", &IOBase::isatty)

.def("readline",

[](IOBase *self, PyTuple *args, PyDict *kwargs) -> PyResult<PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

ASSERT(args);

if (args->elements().empty()) {

return self->readline(-1);

} else if (args->elements().size() > 1) {

return Err(value_error(

"BaseIO.readline expected at most one argument (got {})",

args->elements().size()));

} else {

return PyObject::from(args->elements()[0])

.and_then([self](auto *limit) -> PyResult<PyObject *> {

if (!as<PyInteger>(limit)) { return Err(type_error("")); }

return self->readline(as<PyInteger>(limit)->as_i64());

});

}

})

.def("readlines",

[](IOBase *self, PyTuple *args, PyDict *kwargs) -> PyResult<PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

ASSERT(args);

if (args->elements().empty()) {

return self->readline(-1);

} else if (args->elements().size() > 1) {

return Err(value_error(

"BaseIO.readlines expected at most one argument (got {})",

args->elements().size()));

} else {

return PyObject::from(args->elements()[0])

.and_then([self](auto *hint) -> PyResult<PyObject *> {

if (!as<PyInteger>(hint)) { return Err(type_error("")); }

return self->readlines(as<PyInteger>(hint)->as_i64());

});

}

})

.def("seek", &IOBase::seek)

.def("seekable", &IOBase::seekable)

.def("tell", &IOBase::tell)

.def("truncate", &IOBase::truncate)

.def("writable", &IOBase::writable)

.def("writelines",

[](IOBase *self, PyTuple *args, PyDict *kwargs) -> PyResult<PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

ASSERT(args);

if (args->elements().empty()) {

return Err(

value_error("BaseIO.readlines expected one argument (got 0)"));

} else if (args->elements().size() > 1) {

return Err(value_error(

"BaseIO.readlines expected at most one argument (got {})",

args->elements().size()));

} else {

return PyObject::from(args->elements()[0])

.and_then([self](auto *lines) -> PyResult<PyObject *> {

return self->writelines(lines);

});

}

})

.def("_checkSeekable", &IOBase::check_seekable_)

.def("_checkReadable", &IOBase::check_readable_)

.def("_checkWritable", &IOBase::check_writable_)

.def("__enter__", &IOBase::__enter__)

.def("__exit__", &IOBase::__exit__)

.finalize();

}

module->add_symbol(PyString::create("_IOBase").unwrap(), s_io_base);

return s_io_base;

}

protected:

static PyResult<PyObject *> check_seekable(PyObject *self)

{

return self->get_method(PyString::create("seekable").unwrap())

.and_then([](auto *seekable) { return seekable->call(nullptr, nullptr); })

.and_then([](auto *result) -> PyResult<PyObject *> {

if (result != py_true()) {

return Err(unsupported_operation(

PyTuple::create(String{ "File or stream is not seekable." }).unwrap(),

nullptr));

}

return Ok(py_true());

});

}

static PyResult<PyObject *> check_readable(PyObject *self)

{

return self->get_method(PyString::create("readable").unwrap())

.and_then([](auto *readable) { return readable->call(nullptr, nullptr); })

.and_then([](auto *result) -> PyResult<PyObject *> {

if (result != py_true()) {

return Err(unsupported_operation(

PyTuple::create(String{ "File or stream is not readable." }).unwrap(),

nullptr));

}

return Ok(py_true());

});

}

static PyResult<PyObject *> check_writable(PyObject *self)

{

return self->get_method(PyString::create("writable").unwrap())

.and_then([](auto *writable) { return writable->call(nullptr, nullptr); })

.and_then([](auto *result) -> PyResult<PyObject *> {

if (result != py_true()) {

return Err(unsupported_operation(

PyTuple::create(String{ "File or stream is not writable." }).unwrap(),

nullptr));

}

return Ok(py_true());

});

}

{

friend class ::Heap;

private:

RawIOBase() : RawIOBase(s_io_raw_iobase->type()) {}

protected:

RawIOBase(const PyType *type) : IOBase(type) {}

public:

static constexpr std::string_view __doc__ = "Base class for raw binary I/O.";

static PyResult<RawIOBase *> create(const PyType *type)

{

auto &heap = VirtualMachine::the().heap();

auto *result = heap.allocate<RawIOBase>(type);

if (!result) { return Err(memory_error(sizeof(RawIOBase))); }

return Ok(result);

}

static PyResult<PyObject *> __new__(const PyType *type, PyTuple *, PyDict *)

{

return RawIOBase::create(type);

}

PyType *static_type() const override { return s_io_raw_iobase; }

PyResult<PyObject *> read(int64_t n)

{

if (n < 0) {

return get_method(PyString::create("readall").unwrap()).and_then([](PyObject *readall) {

return readall->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

auto bytes_ = PyBytes::create();

if (bytes_.is_err()) return bytes_;

auto *bytes = bytes_.unwrap();

return get_method(PyString::create("readinto").unwrap())

.and_then([bytes](PyObject *readinto) {

return readinto->call(PyTuple::create(bytes).unwrap(), PyDict::create().unwrap());

})

.and_then([](PyObject *res) -> PyResult<int64_t> {

if (!as<PyInteger>(res)) {

return Err(type_error(

"expected readinto to return an int, got '{}'", res->type()->name()));

}

return Ok(as<PyInteger>(res)->as_i64());

})

.and_then([bytes](const int64_t &n) {

const auto &b = bytes->value().b;

std::vector<std::byte> result{ b.begin(), b.begin() + n };

return PyBytes::create(Bytes{ std::move(result) });

});

}

PyResult<PyObject *> readinto(PyObject *) const

{

return Err(not_implemented_error("_RawIOBase.readinto"));

}

PyResult<PyObject *> write(PyObject *) const

{

return Err(not_implemented_error("_RawIOBase.write"));

}

static PyType *register_type(PyModule *module)

{

if (!s_io_raw_iobase) {

s_io_raw_iobase =

klass<RawIOBase>(module, "_RawIOBase", s_io_base)

.def("read",

[](RawIOBase *self, PyTuple *args, PyDict *kwargs) -> PyResult<PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

int64_t n = -1;

if (args && args->elements().size() > 1) {

return Err(value_error(

"_RawIOBase.read expected at most one argument (got {})",

args->elements().size()));

} else if (args && args->elements().size() == 1) {

auto arg0 = PyObject::from(args->elements()[0]);

if (arg0.is_err()) return arg0;

if (!as<PyInteger>(arg0.unwrap()) && arg0.unwrap() != py_none()) {

return Err(

type_error("argument should be integer or None, not '{}'",

arg0.unwrap()->type()->name()));

}

if (arg0.unwrap() != py_none()) {

n = as<PyInteger>(arg0.unwrap())->as_i64();

}

}

return self->read(n);

})

// .def("readall", &RawIOBase::readall)

.def("readinto", &RawIOBase::readinto)

.def("write", &RawIOBase::write)

.finalize();

}

module->add_symbol(PyString::create("_RawIOBase").unwrap(), s_io_raw_iobase);

return s_io_raw_iobase;

}

{

friend class ::Heap;

BufferedIOBase() : BufferedIOBase(s_io_buffered_io_base) {}

protected:

BufferedIOBase(PyType *type) : IOBase(type) {}

public:

static constexpr std::string_view __doc__ =

"Base class for buffered IO objects.\n"

"\n"

"The main difference with RawIOBase is that the read() method\n"

"supports omitting the size argument, and does not have a default\n"

"implementation that defers to readinto().\n"

"\n"

"In addition, read(), readinto() and write() may raise\n"

"BlockingIOError if the underlying raw stream is in non-blocking\n"

"mode and not ready; unlike their raw counterparts, they will never\n"

"return None.\n"

"\n"

"A typical implementation should not inherit from a RawIOBase\n"

"implementation, but wrap one.\n";

static PyResult<BufferedIOBase *> create(const PyType *type)

{

auto &heap = VirtualMachine::the().heap();

auto *result = heap.allocate<BufferedIOBase>(const_cast<PyType *>(type));

if (!result) { return Err(memory_error(sizeof(BufferedIOBase))); }

return Ok(result);

}

static PyResult<PyObject *> __new__(const PyType *type, PyTuple *, PyDict *)

{

return BufferedIOBase::create(type);

}

PyResult<PyObject *> detach() const

{

return Err(unsupported_operation(PyTuple::create(String{ "detach" }).unwrap(), nullptr));

}

PyResult<PyObject *> read(PyTuple *, PyDict *) const

{

return Err(unsupported_operation(PyTuple::create(String{ "read" }).unwrap(), nullptr));

}

PyResult<PyObject *> read1(PyTuple *, PyDict *) const

{

return Err(unsupported_operation(PyTuple::create(String{ "read1" }).unwrap(), nullptr));

}

PyResult<PyObject *> readinto_generic(PyBuffer &buffer, bool readinto1) const

{

const auto method_name = readinto1 ? "read1" : "read";

auto data =

get_method(PyString::create(method_name).unwrap()).and_then([&buffer](PyObject *read) {

return read->call(

PyTuple::create(Number{ buffer.len }).unwrap(), PyDict::create().unwrap());

});

if (data.is_err()) return data;

if (!as<PyBytes>(data.unwrap())) { return Err(type_error("read() should return bytes")); }

auto data_bytes = as<PyBytes>(data.unwrap())->value().b;

const auto len = data_bytes.size();

if (static_cast<int64_t>(len) > buffer.len) {

return Err(

value_error("read() returned too much data: "

"{} bytes requested, {} returned",

buffer.len,

len));

}

std::copy_n(data_bytes.begin(),

data_bytes.size(),

static_cast<std::byte *>(buffer.buf->get_buffer()));

return PyInteger::create(static_cast<int64_t>(len));

}

PyResult<PyObject *> readinto(PyBuffer &buffer) const

{

return readinto_generic(buffer, false);

}

PyResult<PyObject *> readinto1(PyBuffer &buffer) const

{

return readinto_generic(buffer, true);

}

PyResult<PyObject *> write(PyTuple *, PyDict *) const

{

return Err(unsupported_operation(PyTuple::create(String{ "write" }).unwrap(), nullptr));

}

PyType *static_type() const override { return s_io_buffered_io_base; }

static PyType *register_type(PyModule *module)

{

if (!s_io_buffered_io_base) {

s_io_buffered_io_base =

klass<BufferedIOBase>(module, "_BufferedIOBase", s_io_base)

.def("detach", &BufferedIOBase::detach)

.def("read", &BufferedIOBase::read)

.def("read1", &BufferedIOBase::read1)

.def("readinto",

[](BufferedIOBase *self,

PyTuple *args,

PyDict *kwargs) -> py::PyResult<py::PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

if (!args || args->elements().size() != 1) {

return Err(

type_error("_BufferedIOBase.readinto() takes exactly "

"one argument ({} given)",

args->elements().size()));

}

auto arg = PyObject::from(args->elements()[0]);

if (arg.is_err()) return arg;

PyBuffer buffer;

int flags = 1;

return arg.unwrap()

->get_buffer(buffer, flags)

.or_else([arg](auto) -> PyResult<std::monostate> {

return Err(type_error(

"readinto() argument must be read-write bytes-like "

"object, not {}",

arg.unwrap()->type()->name()));

})

.and_then(

[&buffer, arg, self](auto) -> py::PyResult<py::PyObject *> {

if (buffer.is_ccontiguous()) {

return self->readinto(buffer);

} else {

return Err(type_error(

"readinto() argument must be a contiguous "

"buffer, not {}",

arg.unwrap()->type()->name()));

}

});

})

.def("readinto1",

[](BufferedIOBase *self,

PyTuple *args,

PyDict *kwargs) -> py::PyResult<py::PyObject *> {

ASSERT(!kwargs || kwargs->map().empty());

if (!args || args->elements().size() != 1) {

return Err(

type_error("_BufferedIOBase.readinto1() takes exactly "

"one argument ({} given)",

args->elements().size()));

}

auto arg = PyObject::from(args->elements()[0]);

if (arg.is_err()) return arg;

PyBuffer buffer;

int flags = 1;

return arg.unwrap()

->get_buffer(buffer, flags)

.or_else([arg](auto) -> PyResult<std::monostate> {

return Err(type_error(

"readinto1() argument must be read-write bytes-like "

"object, not {}",

arg.unwrap()->type()->name()));

})

.and_then(

[&buffer, arg, self](auto) -> py::PyResult<py::PyObject *> {

if (buffer.is_ccontiguous()) {

return self->readinto(buffer);

} else {

return Err(type_error(

"readinto() argument must be a contiguous "

"buffer, not {}",

arg.unwrap()->type()->name()));

}

});

})

.def("write", &BufferedIOBase::write)

.finalize();

}

module->add_symbol(PyString::create("_BufferedIOBase").unwrap(), s_io_buffered_io_base);

return s_io_buffered_io_base;

}

{

PyObject *raw{ nullptr };

bool ok{ false };

bool detached{ false };

bool readable_{ false };

bool writable_{ false };

bool finalizing{ false };

bool fast_closed_checks{ false };

std::unique_ptr<std::streambuf> buffer;

bool valid_readbuffer() const { return readable_ && buffer && buffer->in_avail() != -1; }

int64_t readahead() const { return valid_readbuffer() ? buffer->in_avail() : 0; }

bool is_initialized() const { return ok && raw; }

bool is_detached() const { return detached; }

PyResult<std::monostate> check_closed(std::string_view err_msg) const

{

if (is_closed() && readahead() == 0) { return Err(value_error(std::string{ err_msg })); }

return Ok(std::monostate{});

}

PyResult<PyObject *> detach()

{

return static_cast<T *>(this)

->get_method(PyString::create("flush").unwrap())

.and_then([](PyObject *flush) {

return flush->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

})

.and_then([this](PyObject *) {

auto *raw = this->raw;

this->raw = nullptr;

this->detached = true;

this->ok = false;

return Ok(raw);

});

}

PyResult<PyObject *> simple_flush() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return raw->get_method(PyString::create("flush").unwrap()).and_then([](PyObject *flush) {

return flush->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> flush_and_rewind();

PyResult<bool> closed() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return raw->get_attribute(PyString::create("closed").unwrap())

.and_then([](PyObject *closed) {

return truthy(closed, VirtualMachine::the().interpreter());

});

}

PyResult<PyObject *> close()

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

// FIXME add lock

auto r = closed();

if (r.is_err()) return Err(r.unwrap_err());

if (r.unwrap()) return Ok(py_none());

if (finalizing) { TODO(); }

auto res =

static_cast<T *>(this)

->get_method(PyString::create("flush").unwrap())

.and_then([](PyObject *flush) -> PyResult<PyObject *> {

return flush->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

res = raw->get_method(PyString::create("close").unwrap())

.and_then([](PyObject *close) -> PyResult<PyObject *> {

return close->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

if (buffer) { buffer = nullptr; }

return res;

}

PyResult<PyObject *> seekable() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return static_cast<const T *>(this)

->raw->get_method(PyString::create("seekable").unwrap())

.and_then([](PyObject *seekable) -> PyResult<PyObject *> {

return seekable->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> writable() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return static_cast<const T *>(this)

->raw->get_method(PyString::create("writable").unwrap())

.and_then([](PyObject *writable) -> PyResult<PyObject *> {

return writable->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> readable() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return static_cast<const T *>(this)

->get_method(PyString::create("readable").unwrap())

.and_then([](PyObject *readable) -> PyResult<PyObject *> {

return readable->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> fileno() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return static_cast<const T *>(this)

->raw->get_method(PyString::create("fileno").unwrap())

.and_then([](PyObject *fileno) -> PyResult<PyObject *> {

return fileno->call(PyTuple::create().unwrap(), PyDict::create().unwrap());

});

}

PyResult<PyObject *> isatty() const

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

return static_cast<const T *>(this)

->raw->get_method(PyString::create("isatty").unwrap())

.and_then([](PyObject *isatty) -> PyResult<PyObject *> {

return isatty->call(nullptr, nullptr);

});

}

PyResult<PyObject *> _dealloc_warn(PyObject *source) const

{

if (this->ok && this->raw) {

// Best-effort warning fired from a deallocation path; any

// exception raised by the user's _dealloc_warn override has

// nowhere useful to land, so deliberately discard.

(void)this->raw->get_method(PyString::create("_dealloc_warn").unwrap())

.and_then([source](PyObject *_dealloc_warn) {

return _dealloc_warn->call(

PyTuple::create(source).unwrap(), PyDict::create().unwrap());

});

}

return Ok(py_none());

}

bool is_closed() const

{

if (!buffer) return false;

if (this->fast_closed_checks) {

TODO();

} else {

return closed().or_else([](auto) { return Ok(false); }).unwrap();

}

}

PyResult<PyObject *> read(int64_t n)

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

if (n < -1) { return Err(value_error("read length must be non-negative or -1")); }

if (is_closed()) { return Err(value_error("read of closed file")); }

if (n == -1) {

return static_cast<T *>(this)->readall();

} else {

auto res = static_cast<T *>(this)->readfast(n);

if (res.is_ok() && res.unwrap() != py_none()) { return res; }

return static_cast<T *>(this)->readgeneric(n);

}

}

PyResult<PyObject *> read1(int64_t n)

{

if (auto err = this->check_initialized(); err.is_err()) return Err(err.unwrap_err());

if (n < 0) {

// TODO: determine actual buffer size

n = 4096;

}

if (is_closed()) { return Err(value_error("read of closed file")); }

if (n == 0) { return PyBytes::create(); }

const auto have = readahead();

if (have > 0) { return static_cast<T *>(this)->readfast(std::min<int64_t>(n, have)); }

Bytes b;

b.b.resize(n);