All posts by David

A mostly professional code monkey, to contact me translate the following "direct from blog to email" at the website Replace spaces with underscores or dashes, whatever is valid. A script will pick up your email, scrub it against a white list, and if your not a spammer I will get an email.

Flask CRUD with sqlalchemy and jinja2 contextfilters

Quick disclaimer, the Flask CRUD thing is not public domain yet and is very volatile.

The project is here
And the outline for the crud thing is in this commit

First is how the crud is currently constructed

class Equipment(CrudAPI):
    def populate(self):
        self.record_cls = db.Equipment
        self.identity = "equipment"
        self.template_form = "equipment_form.j2.html"
        self.template_list = "equipment_list.j2.html"
        self._listColumn("name", magic_field="magic-string")
        self._listColumn("pretty_name", magic_field="magic-string")
        self._addRelationship("category", "name", magic_field="magic-filter")
        self._addRelationship("subcategory", "name", magic_field="magic-filter")

both vars “template_form” and “template_list” are going to be preset once I am certain that the templates can stand on their own with the context vars provided. The “magic-” params and their use are very much magic (eg really toxic) and would recommend ignoring them.

From there the CrudAPI takes over. Skipping ahead to how this relates to context filters. I had this tag mess here in the template

-{%-      for column_name in origin.list_columns -%}
 -{%-          if column_name in origin.magic_columns -%}
 -        {{ cell("", column_name|title, classes=origin.magic_columns[column_name]) -}}
 -{%-          else -%}
 -        {{ cell("", column_name|title) -}}
 -{%          endif %}
 -{%-      endfor %}

and was really not happy with it. So I dived into Flask and Jinja2’s documentation and code to figure out if I could apply Python code inline.

The answer is yes via jinja2’s contextfilters which are not exposed to Flask but can still be used.

def render_header(context, column_name, value="", **kwargs):
    result = ""
    if column_name in context['origin'].magic_columns:
        result = context['cell'](value, column_name.capitalize(), classes=context['origin'].magic_columns[column_name])
        result = context['cell'](value, column_name.capitalize())
    return result

The trick to going from filter to contextfilter is just applying `my_func.contextfilter = True` outside of your functions scope. From there you have access to almost everything (if not everything). The var “origin” is the CrudAPI’s instance passed to the template.

This has opened a lot more opportunities to do clean up. Taking

{% macro data_attributes(data_map, prefix="data-") -%}
    {%- for name, value in data_map.items() -%}
    {{" "}}{{prefix}}{{name}}="{{value}}"
    {%- endfor -%}
{%- endmacro %}
{% macro cell(name, value, classes=None, data_attrs={}) %}
        <span class="{{- ["cell",classes]|join(" ") if classes else "cell" -}}"{{data_attributes(data_attrs)}}>
      {{- caller() if caller else value -}}</span>
  {%- endmacro -%}

and condensing it down to

{% macro cell(name, value, classes=None, data_attrs={}) %}
        <span class="{{- ["cell",classes]|join(" ") if classes else "cell" -}}" {{data_attrs|dict2attrs("data")|safe}}>
    {{- caller() if caller else value -}}</span>
{%- endmacro -%}

via a simple non-context filter

def dict_to_attributes(attributes, prefix=None):
    results = []
    name2dash = lambda *x: "-".join(x)
    format_str = "%s-{}=\"{}\"" % prefix if prefix else "{}=\"{}\""
    for key, value in attributes.items():
        results.append(format_str.format(key, value))
    #TODO disable autoescape
    return " ".join(results)

Just note that at the moment output is still managed by Jinja’s autoescape and I’d rather not shut that off so calls MUST be suffixed with “|safe” as used above.

As for the Crud API, I feel like that is coming along nicely.

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/<int:record_id>               HEAD,OPTIONS,GET,PUT,DELETE
index                          /                                        HEAD,OPTIONS,GET
static                         /static/<path:filename>                  HEAD,OPTIONS,GET

PyQT5 QMediaPlaylist documentation snafu

QMediaPlaylist has a method


with the signature

bool QMediaPlaylist::addMedia(const QMediaContent &content)

And the documentation suggests for c++ that


should work.

BUT in PyQT5


errors out with unexpected QUrl.

So….. you have to do


in python. A bit unwieldy but PySide2 appears to have stalled from an outsiders perspective.

Initially I suspect that QUrl and QMediaContent inherited from some sort of common base class and were split off and its perhaps an untested use case of using QUrl was lost. I ended up making a bandaid in python with

MakePath = lambda x: QtM.QMediaContent(QtCore.QUrl.fromLocalFile(x))

where QtM is `from PyQt5 import QtMultimedia as QtM`

Hotkeys for CKEditor format/styles

Sourced from

function bib_onKey(event){
            var styles = [], config = this.config,
                codex = { }, format_name = undefined;
            codex[CKEDITOR.ALT + 49] = "format_h1";
            codex[CKEDITOR.ALT + 50] = "format_h2";
            codex[CKEDITOR.ALT + 51] = "format_h3";
            format_name = codex[];
            if (format_name !== undefined) {
      "saveSnapshot"); #save state for undo
                //Format codes are not stored anywhere I could find
                // so build them.
                var style = new[format_name]),
                    elementPath = this.elementPath();
                    #Honestly no idea but its needed to work
                    style._.enterMode = config.enterMode;
                #if format is live, kill it or vice versa
                this[style.checkActive( elementPath )
                    ? "removeStyle"
                    : "applyStyle"
                ]( style );
                #another hit on the undo logic.
       'saveSnapshot' );

put that in the call to CKEditor as a custom config parameter

CKEDITOR.inline("your tag name", { on:{key:bib_onKey}})

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.

From a discussion on a python chat board – How to deal with criticism

The poster asked “This may be offtopic but i really want to share this.
Did you ever faced criticism with your code. How to overcome this criticism ?”

And this was my response.

Every so often I go back to something I wrote 10 years ago and I laugh my ass off. Now if 10 years ago someone had done that, it would have been somewhat of a painful experience. People have already mentioned “you are not you’re code” which is mostly true BUT at same time you are your code at this moment in time. When a peer criticizes your code, you need to quickly detach yourself and hear out what they have to say. Sometimes their input is going to be asinine ( “I prefer naming variable after my kid’s” ) but hopefully its going to real value ( “You should implement dash or camel casing, does ‘expertsexchange’ mean Expert Sex change or Experts exchange? “)

Before I give whatever advice on filtering poisonous vs constructive criticism, going to address why you want criticism. If you are a small person, you will be stuck in a small world and only realize the situation when people are selling cars and you’re still crafting buggy whips to a doomed industry ( eg COBOL & mainframes vs Java & server farms ). To improve your craft, yes you need challenging work but you also need to be exposed to different idea’s or you will not grow professionally.

Now as far as handling criticism and deciding if it has value. #1 is that you need to check your emotions and listen. #2 identify the problem they have with your work and clarify so you have concrete examples of what is and is not the problem. #3 Evaluate the value of fixing the problem “Does this improve things for my team and our success” or “Does this make my product better and or more maintainable?”

If the person cannot give concrete examples of #2, tell them you don’t understand. If they cannot do this without resorting to verbal abuse, conversation is over and escalate to supervision or discontinue discourse.

If the person cannot demonstrate #3 ( eg how does using their children’s names make things “better”? ) escalate or discontinue, telling them that you don’t seen an advantage to their proposal.

Finally, emotion’s get you into a fight you may not be able to win or will have costs to your career down the line. One example is a person I found immensely influential to my career and I thought he was the bee’s knees. This person was outspoken and somewhat vitriolic but he was generally right ( or appeared to be ). 5 years down the road, he seems like a “has been” that is constantly verballing threatening to beat up people when they criticize his code or his behavior “I am a MMA fighter, I will kick your ass”. For the most part I think that guy’s career is on its way out as who wants to collaborate or associate with him? Also I am not talking about Linus Torvald… he’s a completely different kind of crazy with a completely different problem.

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.

Project notes

goals: evaluate speed over memory for pypy using stackless/coroutines with ZMQ and twistless ( if its even possible without rebuilding txZMQ ).

Initial goal would be to take and let two stacks fight it out for an hour ( assuming 10x deep othello minimax ).

Interesting way to safely debug multiprocessing python systems

I have one particular “job” that has 3 sub processes moving as fast as humanly possible to build a report. The main slowdown is an external data source which isn’t outright terrible but its not great either. The worst possible outcome is when this thing hangs or misses available work which it was predisposed to do a lot.

Various kill signals usually failed to give me an idea of where the workers were getting hung up and I wasn’t really excited about putting tracer log messages everywhere. Fortunately I have a dbgp enabled IDE and I found this answer on SO.

Taking that I modified it to look like this:

import traceback, signal
#classdef FeedUserlistWorker which is managed by a custom multiprocessing.Pool implementation.
    def Create(cls, feed, year_month = None):
        signal.signal(signal.SIGUSR1, FeedUserlistWorker._PANIC)
            return cls(year_month=year_month, feed=feed).run()
        except Exception as e:
            from traceback import print_exc

the print_exc is there because there isn’t a very reliable bridge to carry Exceptions from child to parent. Flush’s are there because stdout/stderr are buffered in between the parent pool manager.

    def _PANIC(cls, sig, frame):
        from dbgp.client import brk; brk("", 9090)

The only thing that matters is that call to dbgp. Using that tool, I was able to step up the call stack, fire adhoc commands to inspect variables in the stack frame, and find the exact blocking call, which turned out to be the validation/authentication part of boto s3. That turned out to be a weird problem as I had assumed the busy loop/block was in my own code ( eg while True: never break ), fortunately it has an easy fix!msg/boto-users/0osmP0cUl5Y/5NZBfokIyoUJ which resolved the problem as my Pool manager doesn’t mark tasks complete and failures will only cause the lost task to be resumed from the last point of success.

Cassandra 1.2.x – Wide rows are freaky on AWS m1.xlarge

CQL3 is a very nice abstraction to Cassandra but its important to pay attention to what it is doing.

In SQL land, 1 record == 1 row. In Cassandra 1 record == 1 row, but 2+ records can ALSO be on the same row. This has to do with CQL’s partition and primary keys. Your partition key is what decides which row a record belongs to while the primary key is where the record is in a row. If you only have a primary key and no partition key, 1 record == 1 row, but if you have a composite ( partition key, primary key) every record where partition key is the same is going on the same row.

I had a few rows that were ~30GB in size which put stress on nodes using m1.xlarge ( 8GB heap, 300MB new heap size ) with epic Compaction cycles of doom.