Created
September 22, 2024 16:54
-
-
Save jdonszelmann/13881edbe1b865c21e904bef277f02c5 to your computer and use it in GitHub Desktop.
Pipes in Python
This file contains hidden or 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
class Wrapper: | |
def __init__(self, old): | |
self._old = old | |
def __getattribute__(self, attr): | |
if attr == "__call__": | |
return object.__getattribute__(self, attr) | |
elif attr == "__or__": | |
return object.__getattribute__(self, attr) | |
else: | |
return object.__getattribute__(object.__getattribute__(self, "_old"), attr) | |
def __iter__(self, *args, **kwargs): | |
return object.__getattribute__(object.__getattribute__(self, "_old"), "__iter__")(*args, **kwargs) | |
def __repr__(self, *args, **kwargs): | |
return object.__getattribute__(object.__getattribute__(self, "_old"), "__repr__")(*args, **kwargs) | |
def __or__(self, other): | |
return Wrapper(other(self)) | |
def __call__(self, *args, **kwargs): | |
try: | |
return object.__getattribute__(object.__getattribute__(self, "_old"), "__call__")(*args, **kwargs) | |
except TypeError as e: | |
if "argument" in str(e): | |
def defered(x): | |
return object.__getattribute__( | |
object.__getattribute__(self, "_old"), | |
"__call__" | |
)(*args, x, **kwargs) | |
return defered | |
else: | |
raise | |
def modify_iterator_functions(item_dict): | |
for k in item_dict: | |
if k not in ["object", "type"] and "Exception" not in k and "Error" not in k: | |
if hasattr(item_dict[k], "__subclasses__") and k not in ["dict"]: | |
subclasses = tuple(item_dict[k].__subclasses__()) | |
else: | |
subclasses = () | |
without_exception = subclasses | |
if BaseException not in subclasses: | |
subclasses = (BaseException, *subclasses) | |
while True: | |
try: | |
item_dict[k] = type( | |
f"Wrapper of {k}", | |
subclasses, | |
dict(Wrapper.__dict__) | |
)(item_dict[k]) | |
except TypeError as e: | |
if "not an acceptable base type" in str(e): | |
unacceptable_subclass = [ | |
x.strip() | |
for x in str(e).split(" ") | |
if x.strip().startswith("'") and x.endswith("'") | |
][0] | |
subclasses = tuple(s for s in subclasses if s.__name__ not in unacceptable_subclass) | |
continue | |
elif "__new__" in str(e) or "instance lay-out conflict" in str(e): | |
item_dict[k] = Wrapper(item_dict[k]) | |
break | |
elif "metaclass conflict" in str(e) or "(MRO)" in str(e): | |
subclasses = without_exception | |
else: | |
raise | |
else: | |
break | |
modify_iterator_functions(__builtins__.__dict__) | |
def iter(x): | |
return Wrapper(x) | |
x = [1, 2, 3] | |
res = iter(x) | map(lambda x: x + 3) | max() | |
print(res) | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment