if closed:
return
+ if dealloc_warn := getattr(self, "_dealloc_warn", None):
+ dealloc_warn(self)
+
# If close() fails, the caller logs the exception with
# sys.unraisablehook. close() must be called at the end at __del__().
self.close()
else:
return "<{}.{} name={!r}>".format(modname, clsname, name)
+ def _dealloc_warn(self, source):
+ if dealloc_warn := getattr(self.raw, "_dealloc_warn", None):
+ dealloc_warn(source)
+
### Lower-level APIs ###
def fileno(self):
raise
self._fd = fd
- def __del__(self):
+ def _dealloc_warn(self, source):
if self._fd >= 0 and self._closefd and not self.closed:
import warnings
- warnings.warn('unclosed file %r' % (self,), ResourceWarning,
+ warnings.warn(f'unclosed file {source!r}', ResourceWarning,
stacklevel=2, source=self)
- self.close()
def __getstate__(self):
raise TypeError(f"cannot pickle {self.__class__.__name__!r} object")
if not self.closed:
self._stat_atopen = None
try:
- if self._closefd:
+ if self._closefd and self._fd >= 0:
os.close(self._fd)
finally:
super().close()
def newlines(self):
return self._decoder.newlines if self._decoder else None
+ def _dealloc_warn(self, source):
+ if dealloc_warn := getattr(self.buffer, "_dealloc_warn", None):
+ dealloc_warn(source)
+
class StringIO(TextIOWrapper):
"""Text I/O implementation using an in-memory buffer.
self._check_abc_inheritance(io)
def _check_warn_on_dealloc(self, *args, **kwargs):
- f = open(*args, **kwargs)
+ f = self.open(*args, **kwargs)
r = repr(f)
with self.assertWarns(ResourceWarning) as cm:
f = None
r, w = os.pipe()
fds += r, w
with warnings_helper.check_no_resource_warning(self):
- open(r, *args, closefd=False, **kwargs)
+ self.open(r, *args, closefd=False, **kwargs)
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
def test_warn_on_dealloc_fd(self):