Compare commits
2 Commits
4e1d9cd620
...
83c90879f1
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
83c90879f1 | ||
|
|
07979575b3 |
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
|
||||||
@ -27,7 +27,7 @@ class Explainable:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def use_env(path: str = ".env"):
|
def use_env(cls, path: str = ".env"):
|
||||||
env = Environment(env_file_name=path)
|
env = Environment(env_file_name=path)
|
||||||
Explainable.API_KEY = env.get("OPENAPI_KEY")
|
Explainable.API_KEY = env.get("OPENAPI_KEY")
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user