Tag Archives: python

Filtering and ordering by date with SQLAlchemy

You want the extract command which is documented here -> https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.extract

The list of options are generally the same regardless of dialect/SQL server so a reference of types can be seen here for SQLite3 https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/dialects/sqlite/base.py#L1229

A common base extract type arguments to SQL arguments is here ->https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/sql/compiler.py#L299

Extract() is transformed for SQLite3 into `strftime` -> https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/dialects/sqlite/base.py#L1272

The base visitor/transformer for extract is here https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/sql/compiler.py#L2124

Finally, a basic example might be something like MyTable.query.filter(extract('year', MyTable.date_field) == 2022) which would produce something like SELECT ...hell of a lot of columns... FROM MyTable WHERE STRFTIME("%y", MyTable.date_field); 

Slimmed down twisted compatible reloader script

Working on txweb again, I decided to give it a real flask/django style reloader script. Directly inspired by <a href=”
https://blog.elsdoerfer.name/2010/03/09/twisted-twistd-autoreload/ “>this</a> blog post I decided to cut down on the extraneous bits and also change what files it watched.

https://gist.github.com/devdave/05de2ed2fa2aa0a09ba931db36314e3e

DCDB post-mortem

I went into writing DCDB with little or no plan besides building it around dataclasses. The result is a bit rough and precarious.

That said I think I am going to progress onward with making a DCDB2 library that will change a few things. The first would be to completely separate the DCDB tables themselves from the SQL processing logic in a way similar to sqlalchemy’s session system. I do have some other changes in mind, notably a better separation between the ORM domain classes and business logic as well as changes to how relationship’s work.

On the subject of relationship handling. That one would be a bit more complicated as the DCDB2 design idea I had was to use placeholders for the relationship (what does it connect too and in what way), then have the real instrumented handlers created and assigned to a constructed domain class. That last sentence is a bit painful to read which tells me I need to mull that one over a bit more. Regardless, the hack I put together in DCDB was just way too fragile.

Unit-testing sqlalchemy with pytest

@pytest.fixture(scope="function")
def conn(request):
    ts = int(time.time())
    db_path_name = "db"
    db_name = f"{request.function.__name__}.sqlite3"
    filepath = pathlib.Path(__file__).parent / db_path_name / db_name

    LOG.debug(f"Test DB @ {filepath}")
    engine = create_engine(f"sqlite:///{filepath}")
    connection = engine.connect()

    sal2.Base.metadata.drop_all(bind=engine)
    sal2.Base.metadata.create_all(bind=engine)
    factory = scoped_session(sessionmaker(bind=engine))
    sal2.Base.query = factory.query_property()
    session = factory()

    yield ConnResult(connection, session, engine)

    connection.close()
    engine.dispose()

Inside of my “tests” directory I added a “db” directory. Given the logic above, it spawns an entire new database for each test function so that I can go back and verify my database. For someone elses code, you just need to swap out “sal2” with the module name holding your sqlalchemy base and associated model classes. The only thing I wonder about is the issue with create_all. I remember there is a way to bind the metadata object without create_all but damn if I can remember it right now.

Python dataclass database (DCDB)

Why

While I do use sqlalchemy and to some extent peewee for my projects, I slowly got tired of having to relearn how to write SQL when I’ve known SQL since the mid-90’s.

DCDB’s design is also aiming for simplicity and minimal behind the scenes automagical behaviors.   Instead complexity should be added voluntarily and in such a way that it can be traced back.   

Example

import dataclasses as dcs
import dcdb 

@dcs.dataclass()
class Foo:
    name:str
    age:int

db = dcdb.DBConnection(":memory:") # alternatively this can be a file path
db.bind(Foo)
"""
   Bind doesn't change Foo in the local scope but instead
   it creates a new class DCDB_Foo which is stored to the DBConnection in it's 
   table registry.

   Behind the scenes, a table `Foo` is created to the connected database.  No changes to the name are made (eg pluralization). How you wrote your bound dataclasses is almost exactly how it is stored in the sqlite database.

   An exception is that a .id instance property along with DB methods like: update/save, Create, Get, and Select are added to the class definition.

"""
record = db.t.Foo(name="Bob", age="44")
assert record.name == "Bob"
same_record = db.t.Foo.Get("name=?", "Bob")
assert record.age == 44
assert record.id == same_record.id

record.age = 32
record.save()

same_record = db.t.Foo.Get("age=?", 32)
assert record.id == same_record.id
assert same_record.age == 32

same_record.delete()

"""
Note it is important to notice that currently same_record and 
record have the same .id # property but they are different 
instances and copies of the same record with no shared reference.   
Changes to one copy will not reflect with the other.

"""

Github DCDB

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
https://github.com/devdave/wfmastery/tree/revamp_1/wfmastery
And the outline for the crud thing is in this commit https://github.com/devdave/wfmastery/commit/e249895ddc53c0696f59d3def5718e76855af5b9

https://github.com/devdave/wfmastery/blob/revamp_1/wfmastery/crud.py
https://github.com/devdave/wfmastery/blob/revamp_1/wfmastery/views.py
https://github.com/devdave/wfmastery/blob/revamp_1/wfmastery/templates/equipment_list.j2.html

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("id")
        self._listColumn("hidden")
        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.

@App.template_filter("render_header")
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])
    else:
        result = context['cell'](value, column_name.capitalize())


    return result

render_header.contextfilter=True

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={}) %}
        
      {{- caller() if caller else value -}}
  {%- endmacro -%}

and condensing it down to

{% macro cell(name, value, classes=None, data_attrs={}) %}
        
    {{- caller() if caller else value -}}
{%- endmacro -%}

via a simple non-context filter

@App.template_filter("dict2attrs")
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 http://flask.pocoo.org/snippets/117/ 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.

/foo/<int:bar>

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.

@App.cli.command("list_routes")
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 http://flask.pocoo.org/snippets/117/ "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"))
    print("-"*80)
    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

"HOSTED@LOCAL:5000"
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

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):
        self.name = name
        
    def __call__(self, f):
        decoratedCalls[self.name].append(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):
        self.name = name
        
    def __call__(self, f):
        
        @wraps(f)
        def decorator(inst, *args, **kwargs):
            call("%s-pre" % self.name, inst, args, kwargs)            
            retval = f(inst, *args, **kwargs)
            call("%s-post" % self.name, 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 Foo.bar being called and after Foo.ego is called


@Subscribe("__main__.Foo.bar-pre")        
def preBar(inst, *args, **kwargs):
    if not hasattr(inst, "ext_info"):
        inst.ext_info = "Here"
        
@Subscribe("__main__.Foo.ego-post")
def postEgo(inst, *args, **kwargs):
    if hasattr(inst, "ext_info"):
        print "Extended info is ", inst.ext_info

Our test shows….

x = Foo()
x.bar()
x.blah()
x.ich()
x.ego()
x.say("abba", "dabba")

this as output

__main__.Foo.__init__-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
init'd
__main__.Foo.__init__-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.bar-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
bar
__main__.Foo.bar-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.blah-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
blah
__main__.Foo.blah-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.ich-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
ich
__main__.Foo.ich-post (<__main__.__init__ object at 0x02637890>, None) {}
__main__.Foo.ego-pre (<__main__.__init__ object at 0x02637890>, (), {}) {}
lego
__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

http://wehart.blogspot.com/2009/01/python-plugin-frameworks.html

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 example.py


#App level
from txweb import Site, expose
#twisted
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):

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

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

rootFile = lambda filename : abspath(join(dirname(__file__), filename))
        
class Root(object):
    
    @expose
    def index(self, request):
        """
            Will handle both / and /index paths
        """
        return "Hello From Index!"
    
    @expose
    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("README.md"))
    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 root.hello.world() 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 README.md !

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.