-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add Disposable mixin and helpers (#5)
Add the `Disposable` mixin, which serves as a wrapper for managing resource disposal and cleanup, while also providing the disposal status of an entity. Additionally, include a decorator to guarantee that methods that depend on resources will fail gracefully with descriptive error messages if called on disposed of entities.
- Loading branch information
1 parent
299515a
commit b5e2b3a
Showing
5 changed files
with
329 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -41,6 +41,7 @@ API Reference | |
:recursive: | ||
|
||
sghi.app | ||
sghi.disposable | ||
sghi.exceptions | ||
sghi.typing | ||
sghi.utils | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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", | ||
] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |