Skip to content

Instantly share code, notes, and snippets.

@jdonszelmann
Created September 22, 2024 16:54
Show Gist options
  • Save jdonszelmann/13881edbe1b865c21e904bef277f02c5 to your computer and use it in GitHub Desktop.
Save jdonszelmann/13881edbe1b865c21e904bef277f02c5 to your computer and use it in GitHub Desktop.
Pipes in Python
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