| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138 |
- # SPDX-FileCopyrightText: 2026 geisserml <geisserml@gmail.com>
- # SPDX-License-Identifier: Apache-2.0 OR BSD-3-Clause
- __all__ = ("AutoCastable", "AutoCloseable", "DEBUG_AUTOCLOSE", "LIBRARY_AVAILABLE", "_safe_debug")
- import os
- import sys
- import enum
- import uuid
- import weakref
- import logging
- logger = logging.getLogger(__name__)
- def _safe_debug(msg): # pragma: no cover
- # try to use os.write() rather than print() to avoid "reentrant call" exceptions on shutdown (see https://stackoverflow.com/q/75367828/15547292)
- try:
- os.write(sys.stderr.fileno(), (msg+"\n").encode())
- except Exception: # e.g. io.UnsupportedOperation
- print(msg, file=sys.stderr)
- class _Mutable:
-
- def __init__(self, value):
- self.value = value
-
- def __repr__(self):
- return f"_Mutable({self.value})"
-
- def __bool__(self):
- return bool(self.value)
- DEBUG_AUTOCLOSE = _Mutable(False)
- LIBRARY_AVAILABLE = _Mutable(False) # set to true on library init
- class _STATE (enum.Enum):
- INVALID = -1
- AUTO = 0
- EXPLICIT = 1
- BYPARENT = 2
- class AutoCastable:
-
- @property
- def _as_parameter_(self):
- # trust in the caller not to invoke APIs on an object after .close()
- # if not self.raw:
- # raise RuntimeError("bool(obj.raw) must evaluate to True for use as C function parameter")
- return self.raw
- def _close_template(close_func, raw, obj_repr, state, parent, args, kwargs):
-
- if DEBUG_AUTOCLOSE: # pragma: no cover
- _safe_debug(f"Close ({state.value.name.lower()}) {obj_repr}")
-
- if not LIBRARY_AVAILABLE: # pragma: no cover
- _safe_debug(f"-> Cannot close object; pdfium library is destroyed. This may cause a memory leak.")
- return
-
- assert state.value != _STATE.INVALID
- assert parent is None or not parent._tree_closed()
- close_func(raw, *args, **kwargs)
- class AutoCloseable (AutoCastable):
-
- def __init__(self, close_func, *args, obj=None, needs_free=True, **kwargs):
-
- # proactively prevent accidental double initialization
- assert not hasattr(self, "_finalizer")
-
- self._close_func = close_func
- self._obj = self if obj is None else obj
- self._ex_args = args
- self._ex_kwargs = kwargs
- self._autoclose_state = _Mutable(_STATE.AUTO)
- self._uuid = uuid.uuid4() if DEBUG_AUTOCLOSE else None
-
- self._finalizer = None
- self._kids = []
- if needs_free:
- self._attach_finalizer()
-
-
- def __repr__(self):
- identifier = hex(id(self)) if self._uuid is None else self._uuid.hex[:14]
- return f"<{type(self).__name__} {identifier}>"
-
-
- def _attach_finalizer(self):
- # NOTE this function captures the value of the `parent` property at finalizer installation time
- assert self._finalizer is None
- self._finalizer = weakref.finalize(self._obj, _close_template, self._close_func, self.raw, repr(self), self._autoclose_state, self.parent, self._ex_args, self._ex_kwargs)
-
- def _detach_finalizer(self):
- self._finalizer.detach()
- self._finalizer = None
-
- def _tree_closed(self):
- if self.raw is None:
- return True
- if self.parent != None and self.parent._tree_closed():
- return True
- return False
-
- def _add_kid(self, k):
- self._kids.append( weakref.ref(k) )
-
-
- def close(self, _by_parent=False):
-
- # TODO remove object from parent's kids cache on finalization to avoid unnecessary accumulation
-
- if not self.raw:
- return False
- if not self._finalizer:
- self.raw = None
- return False
-
- for k_ref in self._kids:
- k = k_ref()
- if k and k.raw:
- k.close(_by_parent=True)
-
- self._autoclose_state.value = _STATE.BYPARENT if _by_parent else _STATE.EXPLICIT
- self._finalizer()
- self._autoclose_state.value = _STATE.INVALID
- self.raw = None
- self._finalizer = None
- self._kids.clear()
-
- return True
|