Category Archives: python

Flask list routes (rake equivalent).

While working on a pet project I really wanted a rails rake equivalent for Flask.

Googling led to which gave me enough direction to figure out how to make that work with Python 3.5 and Flask 0.12.

The biggest problem I had with that snippet is that it didn’t account for URL variable rules.


as it would blow up because werkzeug & Flask sanitize inputs to ensure they match the expected type.

I started doing some seriously crazy stuff like monkey patching the rules temporarily to make
ALL converters run through a String/Unicode converter. It’s at this point that I noticed in dbgp (symbolic debugger) that it was naturally converting the rules to strings.

def list_routes():
        Roll through Flask's URL rules and print them out
        Thank you to Jonathan Tushman
            And Thank you to Roger Pence
            Sourced "Helper to list routes (like Rail's rake routes)"

        Note that a lot has possibly changed since that snippet and rule classes have a __str__
            which greatly simplifies all of this

    format_str = lambda *x: "{:30s} {:40s} {}".format(*x)#pylint: disable=W0108
    clean_map = defaultdict(list)

    for rule in App.url_map.iter_rules():
        methods = ",".join(rule.methods)
        clean_map[rule.endpoint].append((methods, str(rule),))

    print(format_str("View handler", "HTTP METHODS", "URL RULE"))
    for endpoint in sorted(clean_map.keys()):
        for rule, methods in sorted(clean_map[endpoint], key=lambda x: x[1]):
            print(format_str(endpoint, methods, rule))

Example output

View handler                   HTTP METHODS                             URL RULE
Equipment                      /equipment/                              OPTIONS,POST
Equipment                      /equipment/                              HEAD,OPTIONS,GET
Equipment                      /equipment/               HEAD,OPTIONS,GET,PUT,DELETE
index                          /                                        HEAD,OPTIONS,GET
static                         /static/                  HEAD,OPTIONS,GET

os.walk for pathlib.Path

I needed the pathlib equivalent to os.walk so I went to the source code for os.walk and reimplemented it to use pathlib instead

def path_walk(top, topdown = False, followlinks = False):
         See Python docs for os.walk, exact same behavior but it yields Path() instances instead
    names = list(top.iterdir())

    dirs = (node for node in names if node.is_dir() is True)
    nondirs =(node for node in names if node.is_dir() is False)

    if topdown:
        yield top, dirs, nondirs

    for name in dirs:
        if followlinks or name.is_symlink() is False:
            for x in path_walk(name, topdown, followlinks):
                yield x

    if topdown is not True:
        yield top, dirs, nondirs

yes there is Path.glob(“**”) but I like how os.walk is structured and this passes my unit-tests.

dirs and nondirs are generators out of habit. List comprehensions would likely be better and it’s almost impossible to avoid the overhead memory cost so having them as is, is just personal preference.

False positive blocking calls in twisted with Chrome

I have a simple txWeb service with endpoints:


Index prints

"hello world %s" % time.time()

Say pushes

"hello %s" % time.time()

onto a 0MQ pub/sub socket

Hear waits for a message to happen on the 0MQ pub/sub socket and than publishes it.

Now say I’ve got 4 browser windows open: 1 called index, 2 are blocking on hear, and finally I call say.

I expected both hears to end their blocking state and print the same “hello 1234” message BUT instead the first hear returns while the second one stays blocking.

This took me a bit to debug BUT what happens is that the first /hear blocks on its call to the server while the second is QUEUED INSIDE CHROME and never calls the server. It’s only after the first one completes ( timeout or success ) that the second one calls the server.

Adventures in SSH: Agent authentication

As mentioned previously; I’m working on hacking/implementing agent support to Twisted.conch. Fortunately, Exarkun pointed me in the direction of twisted.conch.ssh.agent.SSHAgentClient which implements the wire protocol logic of communicating with the Agent, but there is still a gaping hole to fill in.

Briefly, when a user configures their ssh client to allow for agent forwarding, almost immediately after userauth completion, the client sends a session request for ‘’. For openssh, the service then kicks off a process of creating a named socket that usually resides in /tmp/ , announces the user’s agent presence in the shell environment, and then binds a specialized TCP port forwarding channel from the named socket back to the client on a channel called “auth agent”. When a service local ssh client then begins it’s own authentication process, it finds this special socket and sends down the wire a request identities or sign request Agent protocol message. Ideally the response will be a correctly counter-signed value and the user can progress.

The last point can be found in the session.c file of OpenSSH as:

239		nc = channel_new("auth socket",
240		    SSH_CHANNEL_AUTH_SOCKET, sock, sock, -1,
242		    0, "auth socket", 1);

Unfortunately I haven’t hunted down what the global const SSH_CHANNEL_AUTH_SOCKET correlates to in regards to Python. I believe argument 1 “auth socket” is equivalent to the class attribute name in channel.SSHChannel. So the skeleton for an “Auth socket” channel might look like:

class AuthAgentChannel(SSHListenForwardingFactory):
   name = "auth agent"

Alas I haven’t had time to test. I’m debating hacking up some sort of SSH/twisted.conch testing platform to allow for me to execute arbitrary calls, that would probably make this exercise a tad easier to figure out.

The Python data model, an analogy

Given something like:

class Foo(object):

Define Foo.__init__ to determine how to define the interior of your instance.
Define Foo.__new__ to control how you instantiate Foo
Define a metaclass type.__new__ when you want to define what Foo is at run time.

Or put another way:
__init_ decides what the inside of the house looks like, Foo.__new__ decides how you enter your new house or if you even enter a new house at all. And then Type.__new__ and company let you decide what exactly Foo should look like at run time, before you even start instantiating a new Foo.

Stupid until proven otherwise

Feels like a life time ago, I was in the architect’s seat for a multi-million dollar proposed project. From my perspective I made several really damning design choices that I can only argue originated from my lack of more experience. Regardless of these mistakes, the team lead and the other code monkeys got the project off the floor and handling several thousands of concurrent requests/second without to much pain and suffering. So in the grand scheme I suppose the whole thing could be check on the plus column as a success.

Still, years later, my biggest mistake was abusing the crap out of data hiding in the form of protected/private parameters, guarded methods, and such… all because I didn’t trust the junior and mid-level dev’s to get it right. Because of these somewhat draconian design, the natural result was that the code monkeys worked around my restrictions at the cost of precious CPU cycles, extra memory, and some truly horrendous hacks. Sure as the architect I should have caught all of these things, but by that time I had become overbooked as lead on more critical projects while also being the acting DBA and sys. admin for everything ( except the obligatory exchange server which I successfully pretended didn’t exist ).

Thinking over how I would have handled things better, I think the over-arching gist of Alex Martelli’s presentation on API Anti-patterns from PyCon 2011 –

metaclass service bus decorator concept

While playing around with some idea’s for making PyProxy completely overridable ( and also potentially undebuggable ) I started playing around with metaclasses.

Below is from my toxic directory [ gist url ]

from collections import defaultdict
from functools import wraps

Just some preliminary basic utilities

A simple bus implementation, decoratedCalls is a dictionary of service calls, with a list
of callbacks to each service call hook.

decoratedCalls = defaultdict(list)

def call(name, *args, **kwargs):
    print name, args, kwargs
    if name in decoratedCalls:
        for cb in decoratedCalls[name]:
            cb(*args, **kwargs)

Syntactic suger to add clarity later as to what functions are being bound to what. Adding in
debug/logging hooks here could allow for easier tracing of what is called for what methods.

class Subscribe(object):
    def __init__(self, name): = name
    def __call__(self, f):
        return f

Here’s the actual bus decorator, very simple just wraps the decorated function with a pre and post bus calls.

class BusDecorator(object):
    def __init__(self, name): = name
    def __call__(self, f):
        def decorator(inst, *args, **kwargs):
            call("%s-pre" %, inst, args, kwargs)            
            retval = f(inst, *args, **kwargs)
            call("%s-post" %, inst, retval)            
            return retval
        return decorator

And here’s my Bus Metaclass that combines most of the above.

The ease of wrapping the target class is accomplished by cdict which is a dictionary of
every defined attribute of the target class. As you can see it’s trivial to spin
through and decorate every callable with the BusDecorator

class BusWrap(type):
    def __new__(mcs, clsname, bases, cdict):

        modName = cdict.get("__module__", "unknownclass")
        for name in cdict.keys():
            prefix = "%s.%s.%s" % ( modName, clsname, name)
            if callable(cdict[name]):                
                cdict[name] = BusDecorator(prefix)(cdict[name])
        return type.__new__(mcs, name, bases, cdict)

Now give a dirt simple class like

class Foo(object):
    __metaclass__ = BusWrap
    def __init__(self):
        print "init'd"
    def bar(self):
        print "bar"
    def blah(self):
        print "blah"
    def ich(self):
        print "ich"
    def ego(self):
        print "lego"
    def say(self, *args):
        print "Saying ", args

And two service handlers to pre being called and after Foo.ego is called

def preBar(inst, *args, **kwargs):
    if not hasattr(inst, "ext_info"):
        inst.ext_info = "Here"
def postEgo(inst, *args, **kwargs):
    if hasattr(inst, "ext_info"):
        print "Extended info is ", inst.ext_info

Our test shows….

x = Foo()
x.say("abba", "dabba")

this as output

__main__.Foo.__init__-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
__main__.Foo.__init__-post (<__main__.__init__ object at 0x02637890>, None) {} (<__main__.__init__ object at 0x02637890>, (), {}) {}
bar (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.blah-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
__main__.Foo.blah-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.ich-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
__main__.Foo.ich-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.ego-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
__main__.Foo.ego-post (<__main__.__init__ object at 0x02637890>, None) {}
Extended info is  Here
__main__.Foo.say-pre (<__main__.__init__ object at 0x02637890>, ('abba', 'dabba'), {}) {}
Saying  ('abba', 'dabba')
__main__.Foo.say-post (<__main__.__init__ object at 0x02637890>, None) {}

Plugin’s for python

As the #1 google result for “python plugin”, the linked to blog post is extremely valuable for jump starting research into implementing your own plugin system ( like me ) or finding one that is viable for implementation in your project ( possibly like me ).

These resources highlight one reason why there is not a standard Python plug-in framework: there are a variety of different capabilities that a user may want, and the complexity of the framework generally increases as these new capabilities are added

TxWeb Alpha – A different spin on twisted.web

I’ve spent some more time on my current pet, txweb, and I think it’s pretty much at the as good as it gets stage.

Below is the source for the

#App level
from txweb import Site, expose
from twisted.web import server, resource
from twisted.internet import reactor
from twisted.web.static import File

from os.path import abspath, dirname, join

Mostly pretty standard imports for a twisted.web application.

Now here is the “Controllers”, they’re stripped down to bare-bones just to keep it simple

class PageOne(object):

    def foo(self, request):
        return "Hello From PageOne Foo!"        
    def delayed(self, request):
        def delayedResponse():
            request.write("I was delayed :( ")
        reactor.callLater(5, delayedResponse)
        return server.NOT_DONE_YET
class PageTwo(object):

    def index(self, request):
        """ /pagetwo/index """
        return "Hello From PageTwo index!"

rootFile = lambda filename : abspath(join(dirname(__file__), filename))
class Root(object):
    def index(self, request):
            Will handle both / and /index paths
        return "Hello From Index!"
    def __default__(self, request):
            Unless overriden further down, this will catch all 404's
        return "I Caught %s " % request.path
    pageone = PageOne()
    pagetwo = PageTwo()
    readme  = File(rootFile(""))
    license = File(rootFile("txweb/LICENSE.txt"))

Basically a txWeb enabled twisted service converts a URL path to an Object path.

So /hello/world could resolve to if such a construct was provided.

Much more importantly, with the above example, /license resolves to the local file txweb/LICENSE and /readme resolves to !

In summary txweb doesn’t throw away the epic amount of work the Twisted developers and volunteers have put forth, it just presents it in another way.

A friendlier asynchronous twisted web, the ghetto monkey patch way

UPDATE to the UPDATE – A cleaned up and more coherent example of txweb is here
UPDATE – Github repo here

I like twisted, and I like Cherrypy, unfortunately just like my militant atheist friends and my more spiritual friends neither seems to get along with the other.

What to do? MONKEY PATCH + GHETTO HACKING to the rescue!

Note, this is just a mockup of CherryPy’s routing system and not a bridge or interface to CherryPy. There is no CherryPy to be had here, just ghetto py.

from twisted.web import server, resource
from twisted.internet import reactor

def expose(func): = True
    return func

class PageOne(object):

    def foo(self, request):
        return "Hello From PageOne Foo!"    = True
    def delayed(self, request):
        def delayedResponse():
            request.write("I was delayed :( ")
        reactor.callLater(5, delayedResponse)
        return server.NOT_DONE_YET
class PageTwo(object):

    def index(self, request):
        return "Hello From PageTwo index!"
class Root(object):
    def index(self, request):
        return "Hello From Index!"
    def __default__(self, request):
        return "I Caught %s " % request.path
    pageone = PageOne()
    pagetwo = PageTwo()
class OneTimeResource(resource.Resource):
        Monkey patch to avoid rewriting more of twisted's lower web
        layer which does a fantastic job dealing with the minute details
        of receiving and sending HTTP traffic.
        func is a callable and exposed property in the Root OO tree
    def __init__(self, func):
        self.func = func
    def render(self, request):
        #Here would be a fantastic place for a pre-filter
        return self.func(request)
        #ditto here for a post filter
class OverrideSite(server.Site):
        A monkey patch that short circuits the normal
        resource resolution logic @ the getResourceFor point
    def checkAction(self, controller, name):
            On success, returns a bound method from the provided controller instance
            else it return None
        action = None
        if hasattr(controller, name):
                action = getattr(controller, name)
                if not callable(action) or not hasattr(action, "exposed"):
                    action = None
        return action
    def routeRequest(self, request):
        action = None
        response = None
        root = parent = self.resource
        defaultAction = self.checkAction(root, "__default__")
        path = request.path.strip("/").split("/")
        for i in range(len(path)):
            element = path[i]
            parent = root
            root = getattr(root, element, None)
            if root is None:                
            if self.checkAction(root, "__default__"):
                #Check for a catchall default action
                defaultAction = self.checkAction(root, "__default__")
            if element.startswith("_"):
                #500 simplistic security check
                action = lambda request: "500 URI segments cannot start with an underscore"
            if callable(root) and hasattr(root, "exposed") and == True:
                action = root
                request.postpath = path[i:] 
            if action is None:
                if root is not None and self.checkAction(root, "index"):
                    action = self.checkAction(root, "index")
        #action = OneTimeResource(action) if action is not None else OneTimeResource(lambda request:"500 Routing error :(")
        if action is None:
            if defaultAction:
                action = defaultAction
                action = lambda request:"404 :("
        return OneTimeResource(action)         

    def getResourceFor(self, request):
        return self.routeRequest(request)
    Twisted thankfully doesn't do any type checking, so a
    dumb OO graph is A-Okay here.  It will be assigned to
dumb = OverrideSite(Root())

reactor.listenTCP(80, dumb )

Slapped this together in about 30 minutes… so there is a HIGH probability that it is almost entirely edge cased! Still it does work ( for me ) and it doesn’t hijack too much of twisted’s core, so it could be viable with a lot of unit-testing love, some additional sanity checking logics, and maybe some well thought out refactoring.