101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""
|
||
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
|