CODEWITHSUNDEEP
pythonjsonsqlalchemy
David Wolever
David Wolever

Reputation: 154454

Tool for generating SQLAlchemy queries from JSON-esque values?

I'm working on an application that does reporting-type things, and I often need to take a filter specified with JSON, convert that to an SQLAlchemy query, then send the results back to the browser (for example the filter start_date: 1234, end_date: 5678, widget_ids: [1, 2, 3] needs to be converted into the query … WHERE start_date >= 1234 AND end_date <= 5678 AND widget_id in (1, 2, 3)).

Is there a tool which will do this kind of conversion automatically (for example, using suffixes like Django's ORM: start_date__ge: 1234, end_date__le: 5678, widget_id__in: [1, 2, 3])?

Obviously there would be security and performance implications of such a scheme… But I'd like to know if anything like this exists before I build one myself.

Edit: I realize that I could build my own thing, but I'm specifically wondering if there are existing tools/libraries, so I don't need to re-invent the wheel.

Upvotes: 2

Views: 1053

Answers (3)

erilem
erilem

Reputation: 2664

I have some simple code that may be of interest (source):

def create_attr_filter(request, mapped_class):
    """Create an ``and_`` SQLAlchemy filter (a ClauseList object) based
    on the request params (``queryable``, ``eq``, ``ne``, ...).

    Arguments:

    request
        the request.

    mapped_class
        the SQLAlchemy mapped class.
    """

    mapping = {
        'eq'   : '__eq__',
        'ne'   : '__ne__',
        'lt'   : '__lt__',
        'lte'  : '__le__',
        'gt'   : '__gt__',
        'gte'  : '__ge__',
        'like' : 'like',
        'ilike': 'ilike'
    }
    filters = []
    if 'queryable' in request.params:
        queryable = request.params['queryable'].split(',')
        for k in request.params:
            if len(request.params[k]) <= 0 or '__' not in k:
                continue
            col, op = k.split("__")
            if col not in queryable or op not in mapping.keys():
                continue
            column = getattr(mapped_class, col)
            f = getattr(column, mapping[op])(request.params[k])
            filters.append(f)
    return and_(*filters) if len(filters) > 0 else None

Upvotes: 3

Keith
Keith

Reputation: 43024

I wrote something like that. I call it Python proxy. It provides a Javascript API to do RPC to Python, via JSON.

It's part of my open source project.

The pertinent files are:

http://code.google.com/p/pycopia/source/browse/trunk/WWW/pycopia/WWW/json.py#135

Javascript side:

http://code.google.com/p/pycopia/source/browse/trunk/WWW/media/js/proxy.js

Usage:

http://code.google.com/p/pycopia/source/browse/trunk/storage/pycopia/db/webservice.py

Sample configuration:

http://code.google.com/p/pycopia/source/browse/trunk/storage/etc/storage.conf.example

However, it's part of a complete framework, alas not well documented. Requires a recent Linux platform and some setup.

But maybe you can get some ideas from it.

Here is some other code that builds a query from a CLI style (argv) string. It does the dynamcic operator selection.

def _get_query(self, argv):
    mapper = models.class_mapper(self._obj)
    args, kwargs = _query_args(argv[1:], self._environ)
    q = _session.query(self._obj)
    if args:
        grps, left = divmod(len(args), 3)
        if grps:
            for name, op, val in _by_three(args[:grps*3]):
                col = getattr(self._obj, name)
                opm = {"=": col.__eq__, 
                        ">": col.__gt__, 
                        "<": col.__lt__, 
                        "match": col.match, 
                        "contains": col.contains, 
                        "in": col.in_, 
                        "like": col.like}.get(op)
                if opm:
                    if op == "like":
                        val = val.replace("*", "%")
                        val = val.replace(".", "_")
                        if "%" not in val:
                            val = "%" + val + "%"
                    if op == "in":
                        val = val.split(",")
                    q = q.filter(opm(val))
        for name in args[grps*3:]:
            if name.startswith("="):
                q = q.order_by(name[1:])
    if kwargs:
        for name, value in kwargs.items():
            col = getattr(self._obj, name)
            value = CLI.clieval(value)
            q = q.filter(col.__eq__(value))
    return q

That might get you started, also.

Upvotes: 2

van
van

Reputation: 76962

Looking at your sample, your JSON string does not really contain operators, so it is not clear whether it is one of ==, >=, <=, IN, etc. Your JSON string can be easily converted into a dict.

For simple cases where you have ==, you can simply use filter_by providing a named dict:

query_dict = {'name': 'parent-2', 'description': 'test', }
query = query.filter_by(**query_dict)

Similarly, you can just build your query using getattr as answered in SQLAlchemy - build query filter dynamically from dict, but this still needs to know the operator of the condition. In the code below only like is used:

q = session.query(myClass)
for attr, value in web_dict.items():
    q = q.filter(getattr(myClass, attr).like("%%%s%%" % value))

If you need to dynamically specify the operator, then you can further use the getattr:

klass, attr, oper, value = MyClass, "startDate", "__ge__", 1234
q = q.filter(getattr(getattr(klass, attr), oper)(value))

Upvotes: 1

Related Questions