feat: add Disposable mixin and helpers

docs/conf.py

@@ -43,7 +43,7 @@
     ),
 }
 
-autodoc_member_order = 'groupwise'
+autodoc_member_order = "groupwise"
 
 autoapi_python_use_implicit_namespaces = True
 
@@ -56,6 +56,11 @@
 
 nitpick_ignore = [
     ("py:class", "_CT_contra"),  # type annotation not available at runtime
+    ("py:class", "_DE"),  # type annotation only available when type checking
+    ("py:class", "_DT"),  # type annotation only available when type checking
+    ("py:class", "_P"),  # type annotation only available when type checking
+    ("py:class", "_RT"),  # type annotation only available when type checking
+    ("py:class", "TracebackType"),  # Used as type annotation. Only available when type checking
     ("py:class", "concurrent.futures._base.Executor"),  # sphinx can't find it
     ("py:class", "concurrent.futures._base.Future"),  # sphinx can't find it
     ("py:class", "sghi.disposable.decorators._D"),  # private type annotations

docs/index.rst

@@ -41,6 +41,7 @@ API Reference
    :recursive:
 
      sghi.app
+     sghi.disposable
      sghi.exceptions
      sghi.typing
      sghi.utils

pyproject.toml

@@ -99,7 +99,10 @@ exclude_lines = [
     '@(abc\.)?abstractmethod',
 
     # Don't complain about conditional TYPE_CHECKING blocks:
-    "if TYPE_CHECKING:",
+    'if (typing\.)?TYPE_CHECKING:',
+
+    # Don't complain about overloads:
+    '@(typing\.)?overload',
 ]
 show_missing = true
 

src/sghi/disposable/__init__.py

@@ -0,0 +1,212 @@
+"""
+Resource disposal mixin, checks and other helpers.
+"""
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+from contextlib import AbstractContextManager
+from functools import wraps
+from typing import TYPE_CHECKING, Concatenate, ParamSpec, TypeVar, overload
+
+from ..exceptions import SGHIError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+    from types import TracebackType
+
+
+# =============================================================================
+# TYPES
+# =============================================================================
+
+
+_DE = TypeVar("_DE", bound="ResourceDisposedError")
+_DT = TypeVar("_DT", bound="Disposable")
+_P = ParamSpec("_P")
+_RT = TypeVar("_RT")
+
+
+# =============================================================================
+# EXCEPTIONS
+# =============================================================================
+
+
+class ResourceDisposedError(SGHIError):
+    """Indicates that a :class:`Disposable` item has already been disposed."""
+
+    def __init__(self, message: str | None = "Resource already disposed."):
+        """
+        Initialize a new instance of `ResourceDisposedError`.
+
+        :param message: Optional custom error message. If not provided, a
+            default message indicating that the resource is already disposed
+            will be used.
+        """
+        super().__init__(message=message)
+
+
+# =============================================================================
+# DISPOSABLE MIXIN
+# =============================================================================
+
+
+class Disposable(AbstractContextManager, metaclass=ABCMeta):
+    """An entity that uses resources that need to be cleaned up.
+
+    As such, this interface supports the
+    :doc:`context manager protocol<python:library/contextlib>` making its
+    derivatives usable with Python's ``with`` statement. Implementors should
+    override the :meth:`dispose` method and define the resource clean up logic.
+    The :attr:`is_disposed` property can be used to check whether an instance
+    has been disposed.
+    """
+
+    __slots__ = ()
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
+        """Exit the context manager and call the :meth:`dispose` method.
+
+        :param exc_type: The type of the exception being handled. If an
+            exception was raised and is being propagated, this will be the
+            exception type. Otherwise, it will be ``None``.
+        :param exc_val: The exception instance. If no exception was raised,
+            this will be ``None``.
+        :param exc_tb: The traceback for the exception. If no exception was
+            raised, this will be ``None``.
+
+        :return: `False`.
+        """
+        super().__exit__(exc_type, exc_val, exc_tb)
+        self.dispose()
+        return False
+
+    @property
+    @abstractmethod
+    def is_disposed(self) -> bool:
+        """
+        Return ``True`` if this object has already been disposed, ``False``
+        otherwise.
+
+        :return: ``True`` if this object has been disposed, ``False``
+            otherwise.
+        """
+        ...
+
+    @abstractmethod
+    def dispose(self) -> None:
+        """Release any underlying resources contained by this object.
+
+        After this method returns successfully, the :attr:`is_disposed`
+        property should return ``True``.
+
+        .. note::
+            Unless otherwise specified, trying to use methods of a
+            ``Disposable`` instance decorated with the :func:`not_disposed`
+            decorator after this method returns should generally be considered
+            a programming error and should result in a
+            :exc:`ResourceDisposedError` being raised.
+
+            This method should be idempotent allowing it to be called more
+            than once; only the first call, however, should have an effect.
+
+        :return: None.
+        """
+        ...
+
+
+# =============================================================================
+# DECORATORS
+# =============================================================================
+
+
+@overload
+def not_disposed(
+    f: Callable[Concatenate[_DT, _P], _RT],
+    *,
+    exc_factory: Callable[[], _DE] = ResourceDisposedError,
+) -> Callable[Concatenate[_DT, _P], _RT]:
+    ...
+
+
+@overload
+def not_disposed(
+    f: None = None,
+    *,
+    exc_factory: Callable[[], _DE] = ResourceDisposedError,
+) -> Callable[
+    [Callable[Concatenate[_DT, _P], _RT]],
+    Callable[Concatenate[_DT, _P], _RT],
+]:
+    ...
+
+
+def not_disposed(
+    f: Callable[Concatenate[_DT, _P], _RT] | None = None,
+    *,
+    exc_factory: Callable[[], _DE] = ResourceDisposedError,
+) -> Callable[Concatenate[_DT, _P], _RT] | Callable[
+    [Callable[Concatenate[_DT, _P], _RT]],
+    Callable[Concatenate[_DT, _P], _RT],
+]:
+    """Decorate a function with the resource disposal check.
+
+    This decorator ensures a :class:`Disposable` item has not been disposed. If
+    the item is disposed, i.e. the :attr:`~Disposable.is_disposed` property
+    returns ``True``, then an instance of :exc:`ResourceDisposedError` or it's
+    derivatives is raised.
+
+    .. important::
+
+        This decorator *MUST* be used on methods bound to an instance of the
+        ``Disposable`` interface. It requires that the first parameter of the
+        decorated method, i.e. ``self``, be an instance of ``Disposable``.
+
+    :param f: The function to be decorated. The first argument of the
+        function should be a ``Disposable`` instance.
+    :param exc_factory: An optional callable that creates instances of
+        ``ResourceDisposedError`` or its subclasses. This is only called
+        if the resource is disposed. If not provided, a default factory
+        that creates ``ResourceDisposedError`` instances will be used.
+
+    :return: The decorated function.
+    """
+    def wrap(
+        _f: Callable[Concatenate[_DT, _P], _RT],
+    ) -> Callable[Concatenate[_DT, _P], _RT]:
+
+        @wraps(_f)
+        def wrapper(
+            disposable: _DT,
+            *args: _P.args,
+            **kwargs: _P.kwargs,
+        ) -> _RT:
+            if disposable.is_disposed:
+                raise exc_factory()
+            return _f(disposable, *args, **kwargs)
+
+        return wrapper
+
+    # Whether `f` is None or not depends on the usage of the decorator. It's a
+    # method when used as `@not_disposed` and None when used as
+    # `@not_disposed()` or `@not_disposed(exc_factory=...)`.
+    if f is None:
+        return wrap
+
+    return wrap(f)
+
+
+# =============================================================================
+# MODULE EXPORTS
+# =============================================================================
+
+
+__all__ = [
+    "Disposable",
+    "ResourceDisposedError",
+    "not_disposed",
+]

test/sghi/disposable_tests.py

@@ -0,0 +1,106 @@
+from unittest import TestCase
+
+import pytest
+
+from sghi.disposable import Disposable, ResourceDisposedError, not_disposed
+
+# =============================================================================
+# DISPOSABLES TEST IMPLEMENTATIONS
+# =============================================================================
+
+
+class _SomeItemDisposedError(ResourceDisposedError):
+
+    def __init__(self, msg: str = "SomeDisposableItem is already disposed"):
+        super().__init__(message=msg)
+
+
+class _SomeDisposableItem(Disposable):
+
+    __slots__ = ("_is_disposed",)
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._is_disposed: bool = False
+
+    @property
+    def is_disposed(self) -> bool:
+        return self._is_disposed
+
+    def dispose(self) -> None:
+        self._is_disposed = True
+
+    @not_disposed
+    def use_resources1(self) -> None:
+        ...
+
+    @not_disposed(exc_factory=_SomeItemDisposedError)
+    def use_resources2(self) -> str:
+        return "Some Results!"
+
+
+# =============================================================================
+# TESTS
+# =============================================================================
+
+
+class TestDisposable(TestCase):
+    """Tests for the :class:`Disposable` mixin."""
+
+    def test_object_is_disposed_on_context_manager_exit(self) -> None:
+        """
+        As per the default implementation of the :class:`Disposable` mixin ,
+        a ``Disposable`` object's ``dispose()`` method should be invoked when
+        exiting a context manager.
+        """
+
+        with _SomeDisposableItem() as disposable:
+            # Resource should not be disposed at this time
+            assert not disposable.is_disposed
+
+        assert disposable.is_disposed
+
+
+class TestNotDisposedDecorator(TestCase):
+    """Tests for the :func:`not_disposed` decorator."""
+
+    def test_decorated_methods_return_normally_when_not_disposed(self) -> None:
+        """
+        Methods decorated using the ``not_disposed`` decorator should return
+        normally, i.e., without raising :exc:`ResourceDisposedError` if their
+        bound ``Disposable`` object is yet to be disposed.
+        """
+        with _SomeDisposableItem() as disposable:
+            # Resource should not be disposed at this time
+            assert not disposable.is_disposed
+
+            try:
+                disposable.use_resources1()
+                assert disposable.use_resources2() == "Some Results!"
+            except ResourceDisposedError:
+                pytest.fail(reason="Resource should not be disposed yet.")
+
+        assert disposable.is_disposed
+
+    def test_decorated_methods_raise_expected_errors_when_disposed(self) -> None:  # noqa: E501
+        """
+        Methods decorated using the ``not_disposed`` decorator should raise
+        :exc:`ResourceDisposedError` or it's derivatives (depending on whether
+        the ``exc_factory`` is provided) when invoked after their bound
+        ``Disposable`` object is disposed.
+        """
+
+        disposable: _SomeDisposableItem = _SomeDisposableItem()
+        disposable.dispose()
+
+        assert disposable.is_disposed
+        with pytest.raises(ResourceDisposedError) as exc_info1:
+            disposable.use_resources1()
+
+        with pytest.raises(_SomeItemDisposedError) as exc_info2:
+            disposable.use_resources2()
+
+        assert exc_info1.type is ResourceDisposedError
+        assert exc_info1.value.message == "Resource already disposed."
+        assert exc_info2.type is _SomeItemDisposedError
+        assert exc_info2.value.message == "SomeDisposableItem is already disposed"  # noqa: E501