bases.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. # SPDX-FileCopyrightText: 2026 geisserml <geisserml@gmail.com>
  2. # SPDX-License-Identifier: Apache-2.0 OR BSD-3-Clause
  3. __all__ = ("AutoCastable", "AutoCloseable", "DEBUG_AUTOCLOSE", "LIBRARY_AVAILABLE", "_safe_debug")
  4. import os
  5. import sys
  6. import enum
  7. import uuid
  8. import weakref
  9. import logging
  10. logger = logging.getLogger(__name__)
  11. def _safe_debug(msg): # pragma: no cover
  12. # try to use os.write() rather than print() to avoid "reentrant call" exceptions on shutdown (see https://stackoverflow.com/q/75367828/15547292)
  13. try:
  14. os.write(sys.stderr.fileno(), (msg+"\n").encode())
  15. except Exception: # e.g. io.UnsupportedOperation
  16. print(msg, file=sys.stderr)
  17. class _Mutable:
  18. def __init__(self, value):
  19. self.value = value
  20. def __repr__(self):
  21. return f"_Mutable({self.value})"
  22. def __bool__(self):
  23. return bool(self.value)
  24. DEBUG_AUTOCLOSE = _Mutable(False)
  25. LIBRARY_AVAILABLE = _Mutable(False) # set to true on library init
  26. class _STATE (enum.Enum):
  27. INVALID = -1
  28. AUTO = 0
  29. EXPLICIT = 1
  30. BYPARENT = 2
  31. class AutoCastable:
  32. @property
  33. def _as_parameter_(self):
  34. # trust in the caller not to invoke APIs on an object after .close()
  35. # if not self.raw:
  36. # raise RuntimeError("bool(obj.raw) must evaluate to True for use as C function parameter")
  37. return self.raw
  38. def _close_template(close_func, raw, obj_repr, state, parent, args, kwargs):
  39. if DEBUG_AUTOCLOSE: # pragma: no cover
  40. _safe_debug(f"Close ({state.value.name.lower()}) {obj_repr}")
  41. if not LIBRARY_AVAILABLE: # pragma: no cover
  42. _safe_debug(f"-> Cannot close object; pdfium library is destroyed. This may cause a memory leak.")
  43. return
  44. assert state.value != _STATE.INVALID
  45. assert parent is None or not parent._tree_closed()
  46. close_func(raw, *args, **kwargs)
  47. class AutoCloseable (AutoCastable):
  48. def __init__(self, close_func, *args, obj=None, needs_free=True, **kwargs):
  49. # proactively prevent accidental double initialization
  50. assert not hasattr(self, "_finalizer")
  51. self._close_func = close_func
  52. self._obj = self if obj is None else obj
  53. self._ex_args = args
  54. self._ex_kwargs = kwargs
  55. self._autoclose_state = _Mutable(_STATE.AUTO)
  56. self._uuid = uuid.uuid4() if DEBUG_AUTOCLOSE else None
  57. self._finalizer = None
  58. self._kids = []
  59. if needs_free:
  60. self._attach_finalizer()
  61. def __repr__(self):
  62. identifier = hex(id(self)) if self._uuid is None else self._uuid.hex[:14]
  63. return f"<{type(self).__name__} {identifier}>"
  64. def _attach_finalizer(self):
  65. # NOTE this function captures the value of the `parent` property at finalizer installation time
  66. assert self._finalizer is None
  67. 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)
  68. def _detach_finalizer(self):
  69. self._finalizer.detach()
  70. self._finalizer = None
  71. def _tree_closed(self):
  72. if self.raw is None:
  73. return True
  74. if self.parent != None and self.parent._tree_closed():
  75. return True
  76. return False
  77. def _add_kid(self, k):
  78. self._kids.append( weakref.ref(k) )
  79. def close(self, _by_parent=False):
  80. # TODO remove object from parent's kids cache on finalization to avoid unnecessary accumulation
  81. if not self.raw:
  82. return False
  83. if not self._finalizer:
  84. self.raw = None
  85. return False
  86. for k_ref in self._kids:
  87. k = k_ref()
  88. if k and k.raw:
  89. k.close(_by_parent=True)
  90. self._autoclose_state.value = _STATE.BYPARENT if _by_parent else _STATE.EXPLICIT
  91. self._finalizer()
  92. self._autoclose_state.value = _STATE.INVALID
  93. self.raw = None
  94. self._finalizer = None
  95. self._kids.clear()
  96. return True