feat: implement Chainable class with chaining support and custom exceptions
This commit is contained in:
parent
07979575b3
commit
83c90879f1
100
multinut/chainable.py
Normal file
100
multinut/chainable.py
Normal file
@ -0,0 +1,100 @@
|
||||
"""
|
||||
chainable.py – A lightweight fluent interface helper for Python.
|
||||
|
||||
Provides:
|
||||
- Chainable base class
|
||||
- Decorators: @chain, @final_chain
|
||||
- Custom exceptions for clear error handling
|
||||
"""
|
||||
|
||||
# === Exceptions ===
|
||||
class ChainException(Exception): pass
|
||||
class ChainLockedError(ChainException): pass
|
||||
class ChainWhitelistError(ChainException): pass
|
||||
class ChainBlacklistError(ChainException): pass
|
||||
class ChainStoppedError(ChainException): pass
|
||||
|
||||
|
||||
# === Core system ===
|
||||
class Chainable:
|
||||
"""Base class for chainable objects with whitelist/blacklist/stop/lock support."""
|
||||
|
||||
def __init__(self):
|
||||
self._chain_whitelist = None
|
||||
self._chain_blacklist = None
|
||||
self._chain_locked = False
|
||||
self._chain_stopped = False
|
||||
|
||||
def _check_chainable(self, func_name, skip_whitelist=False, skip_blacklist=False):
|
||||
if self._chain_locked:
|
||||
raise ChainLockedError(f"⛔ `{func_name}` cannot be chained after a final method.")
|
||||
if self._chain_stopped:
|
||||
raise ChainStoppedError(f"🛑 `{func_name}` cannot be chained, chain stopped.")
|
||||
if not skip_whitelist and self._chain_whitelist and func_name not in self._chain_whitelist:
|
||||
raise ChainWhitelistError(f"❌ `{func_name}` not allowed. Whitelist: {self._chain_whitelist}")
|
||||
if not skip_blacklist and self._chain_blacklist and func_name in self._chain_blacklist:
|
||||
raise ChainBlacklistError(f"🚫 `{func_name}` is blacklisted.")
|
||||
|
||||
def _apply_restrictions(self, whitelist=None, blacklist=None, reset=False):
|
||||
if reset or (whitelist is None and blacklist is None):
|
||||
self._chain_whitelist = None
|
||||
self._chain_blacklist = None
|
||||
else:
|
||||
if whitelist is not None:
|
||||
self._chain_whitelist = [f.__name__ if callable(f) else f for f in whitelist]
|
||||
if blacklist is not None:
|
||||
self._chain_blacklist = [f.__name__ if callable(f) else f for f in blacklist]
|
||||
|
||||
def _lock_chain(self):
|
||||
self._chain_locked = True
|
||||
|
||||
def stop_chain(self, exc: Exception | None = None):
|
||||
"""Stop chaining. Optionally raise an exception immediately."""
|
||||
self._chain_stopped = True
|
||||
if exc:
|
||||
raise exc
|
||||
|
||||
|
||||
# === Decorators ===
|
||||
def chain(whitelist=None, blacklist=None, ignore_whitelist=False, ignore_blacklist=False):
|
||||
"""
|
||||
Mark a method as chainable.
|
||||
|
||||
Args:
|
||||
whitelist (list[str|callable]): Methods allowed to follow.
|
||||
blacklist (list[str|callable]): Methods disallowed to follow.
|
||||
ignore_whitelist (bool): Skip whitelist checks for this method.
|
||||
ignore_blacklist (bool): Skip blacklist checks for this method.
|
||||
"""
|
||||
def decorator(func):
|
||||
def wrapper(self, *args, **kwargs):
|
||||
self._check_chainable(
|
||||
func.__name__,
|
||||
skip_whitelist=ignore_whitelist,
|
||||
skip_blacklist=ignore_blacklist
|
||||
)
|
||||
func(self, *args, **kwargs)
|
||||
|
||||
if ignore_whitelist or ignore_blacklist:
|
||||
self._apply_restrictions(reset=True)
|
||||
elif whitelist is not None or blacklist is not None:
|
||||
self._apply_restrictions(whitelist, blacklist)
|
||||
else:
|
||||
self._apply_restrictions(reset=True)
|
||||
|
||||
return self
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
|
||||
def final_chain(func):
|
||||
"""
|
||||
Mark a method as final in the chain.
|
||||
After calling, chaining is locked.
|
||||
"""
|
||||
def wrapper(self, *args, **kwargs):
|
||||
self._check_chainable(func.__name__)
|
||||
func(self, *args, **kwargs)
|
||||
self._lock_chain()
|
||||
return self
|
||||
return wrapper
|
||||
Loading…
x
Reference in New Issue
Block a user