Skip to content

Instantly share code, notes, and snippets.

@agumonkey agumonkey/fun.py
Last active Mar 9, 2019

Embed
What would you like to do?
pipes in python
import urllib.request as rq
class ConfusedThreader(Exception):
pass
class Fun:
def __getattr__(self,attribute):
def aux(o):
return getattr(o,attribute)()
return aux
fn = Fun()
done = lambda t: t.val()
class Threader:
def __init__(self, v):
self.v = v
self.fns = []
def val(self):
r = self.v
for f in self.fns:
r = f(r)
return r
def __or__(self,other):
if other is done:
return done(self)
elif callable(other):
self.fns.append(other)
return self
elif other is None:
return self.val()
else:
raise ConfusedThreader()
class DebuggingThreader(Threader):
def __init__(self, v, verbose=True):
super().__init__(v)
self.verbose = verbose
self.trace = {}
def val(self):
r = self.v
for f in self.fns:
try:
o = r
r = f(r)
self.trace[f] = (o, r)
if self.verbose:
print('[debug]',f,o,r)
except Exception as e:
print(e)
return r
def Thread(o, debug=False):
return DebuggingThreader(o) if debug else Threader(o)
DEBUG=True
def demo():
Thread("https://www.example.com",DEBUG) \
| rq.urlopen \
| fn.readlines \
| b''.join \
| fn.decode \
| print \
| done
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.