Small DSL: Converting JSON to MSSQL query

Question

Scenario

Once upon a time, there was this human (ahem, me, ahem) who tried (and succeeded) to build a small program which would convert JSON to a MSSQL query using Python 3.x.

---

How does it work?

The user makes a POST request with a JSON (in the code, I've used a JSON file) which looks like this:

{
  "endpoint": "rfc",
  "expression": [
    {
      "field": "first_name",
      "operator": "EQ",
      "value": "Dixie"
    },
    "AND",
    [
      {
        "field": "last_name",
        "operator": "NEQ",
        "value": "Smith"
      },
      "OR",
      {
        "field": "middle_name",
        "operator": "EQ",
        "value": "Sam"
      }
    ]
  ],
  "limit": "2"
}

Now, the key point in this story is the expression which is being translated to a query, using recursion, as follows:

def process(data):
    """
    :param data: JSON Object (dict).
    :return: where clause (str) built from data
    """
    where_clause = ""
    if isinstance(data, list):
        for part in data:
            if part not in LOGICAL_OPERATORS:
                where_clause += " ({}) ".format(process(part))
            else:
                where_clause += process(part)
    elif isinstance(data, dict):
        where_clause += " {} {} '{}'".format(data["field"], COMPARISON_OPERATORS[data["operator"]], data["value"])
    elif isinstance(data, str):
        return data
    return where_clause

The entire code, is as follows:

import json

LOGICAL_OPERATORS = ("AND", "OR")

COMPARISON_OPERATORS = {
    "LT": "<",
    "GT": ">",
    "LTE": "<=",
    "GTE": ">=",
    "EQ": "=",
    "NEQ": "!="
}


def get_data(file):
    """
    :param file: JSON file.
    :return: data as a json object (dict).
    """
    with open(file) as data_file:
        return json.load(data_file)


def process(data):
    """
    :param data: JSON Object (dict).
    :return: where clause (str) built from data
    """
    where_clause = ""
    if isinstance(data, list):
        for part in data:
            if part not in LOGICAL_OPERATORS:
                where_clause += " ({}) ".format(process(part))
            else:
                where_clause += process(part)
    elif isinstance(data, dict):
        where_clause += " {} {} '{}'".format(data["field"], COMPARISON_OPERATORS[data["operator"]], data["value"])
    elif isinstance(data, str):
        return data
    return where_clause


def main():
    expression = get_data("other.json")["expression"]
    where_clause = process(expression)

    return "SELECT * FROM table WHERE {}".format(where_clause)

if __name__ == '__main__':
    print(main())

---

Questions:

• Will the recursive function scale good on a larger expression? Now, that expression can go in depth as long as the memory goes. Should I add a limit to it?
• Did I miss something? Maybe an edge case?
• Is there an already builtin in out there for this?

Answer 1

Escaping and SQL injections

One of the major problems in the presented code is that the code is vulnerable to SQL injection attacks and does not properly sanitize, validate and escape the field values.

For example, if a last_name contains a single quote, for example O'Reilly:

{
    "field": "last_name",
    "operator": "NEQ",
    "value": "O'Reilly"
},

The generated query would be:

SELECT * FROM table WHERE  ( first_name = 'Dixie') AND ( ( last_name != 'O'Reilly') OR ( middle_name = 'Sam') ) 

which is syntactically incorrect query and would raise an error if executed (which may be a sign of not properly sanitized inputs for a potential "bad" user).

Then, we can try:

{
  "endpoint": "rfc",
  "expression": [
    {
      "field": "first_name",
      "operator": "EQ",
      "value": "' or '1' = '1"
    }
  ],
  "limit": "2"
}

which will result into:

SELECT * FROM table WHERE  ( first_name = '' or '1' = '1') 

which will match everything in the table.

Possible to solution to the problem

A possible solution to the problem would be to use parameterized queries and named placeholders where names would come from the field parameter (assuming it's unique). Something along these lines (please test thoroughly):

def process(data, parameters={}):
    """
    :param data: JSON Object (dict). 
    :param parameters: dict.
    :return: where clause (str) built from data
    """
    where_clause = ""
    if isinstance(data, list):
        for part in data:
            if part not in LOGICAL_OPERATORS:
                where_clause += " ({}) ".format(process(part, parameters))
            else:
                where_clause += process(part, parameters)
    elif isinstance(data, dict):
        where_clause += " {} {} %({})s ".format(data["field"], COMPARISON_OPERATORS[data["operator"]], data["field"])
        parameters[data["field"]] = data["value"]
    elif isinstance(data, str):
        return data
    return where_clause


def main():
    expression = get_data("other.json")["expression"]
    parameters = {}
    where_clause = process(expression, parameters)

    return "SELECT * FROM table WHERE {}".format(where_clause), parameters

---

Is there an already builtin in out there for this?

I would look into using ORMs like SQLAlchemy to generate queries. This way you would not have to deal with SQL at all, working with Python abstraction layer only.