from functools import wraps
from random import randint
import time
from klein import Klein
from twisted.internet import defer, reactor
def startMetric(f):
@wraps(f)
def deco(*args, **kw):
req = args[1]
print(f"[x] start:{time.time()} path:{req.path.decode('utf8')}")
result = defer.maybeDeferred(f, *args, **kw)
result.addBoth(endMetric, req)
return result
return deco
def endMetric(result, req):
print(f"[x] end:{time.time()} path:{req.path.decode('utf8')} status:{req.code}")
return result
class MyApp:
rtr = Klein()
@rtr.route("/hello")
@startMetric
def hello(self, req):
d = defer.Deferred()
delay = randint(1, 5)
reactor.callLater(delay, d.callback, f"delay:{delay}")
req.setResponseCode(403, b"whoopsie")
return d
def main():
app = MyApp()
app.rtr.run("localhost", 9999)
main()