pythonsqlfor-loopsqlbuilderpython-sql

Creating an SQL where condition dynamically from a list of dicts


I am currently working on a requirement where I need to build a SQL from a JSON that has details like TABLE_NAME, columns to select, and conditions for the where clause. I am using the sql package in python to build the query as below:

import sql
class QueryGetter(sql.FromItem):

    def __init__(self, table_name: str):
        self.table = table_name

    def get_columns(self, input_columns: List):
        return [sql.Column(self, col) for col in input_columns]

    # I need help with the implementation of this method.
    # As an example of how the condition is built, a hardcoded condition is built in main method below
    def get_cond(self, filters: List[dict]):
        pass
    
    # This helps return the parameters tuple to replace %s in the query
    @property
    def params(self):
        return ()

    # This helps return the string "FROM TABLE" when the query object is printed
    def __str__(self):
        return self.table


if __name__ == "__main__":
    input = {
        "TABLE_NAME" : "TABLE"
        "COLS": [
            "COL1",
            "COL2",
        ]
        "CONDS":[
            {
                "COND_COL": "COL3",
                "COND_OP": "=",
                "COND_VAL": "VAL1"
            },
            {
                "COND_COL": "COL4",
                "COND_OP": "<",
                "COND_VAL": "VAL2"
            },
            {
                "COND_COL": "COL5",
                "COND_OP": "!=",
                "COND_VAL": "VAL3"
            },
            {
                "COND_COL": "COL6",
                "COND_OP": ">",
                "COND_VAL": "VAL4"
            }
        ]
    }
    obj = QueryGetter(input["TABLE_NAME"])
    cols = obj.get_columns(input["COLS"])
    
    # The below statement is equivalent of SELECT COL1, COL2 FROM TABLE
    stmt = obj.select(*cols)
    
    # Create hardcoded columns in where clause
    cond_col1 = sql.Column(obj, input["CONDS"][0]["COND_COL"]
    cond_col2 = sql.Column(obj, input["CONDS"][1]["COND_COL"]
    cond_col3 = sql.Column(obj, input["CONDS"][2]["COND_COL"]
    cond_col4 = sql.Column(obj, input["CONDS"][3]["COND_COL"]
    # Hardcoded condition below that I need to generate with obj.get_cond() method
    # Below statement is based on each item in the list input["CONDS"]
    cond = (cond_col1 == "VAL1") & (cond_col2 < "VAL2") & (cond_col3 != "VAL3") & (cond_col4 > "VAL4")

    # Add this cond object to the SELECT statement prepared earlier
    stmt.where = cond
    
    # After making all the connections to database that is being skipped here, the prepared statement can be executed as below
    rows = cursor.execute(stmt, stmt.params).fetchall()

The requirement is that I shouldn't use raw SQL statements and I must do it this way only. I have done everything else except iterating thru the list of dicts and dynamically create a condition from it. Any help is much appreciated.

Thanks in anticipation!


Solution

  • Use the operator module to convert condition names to functions, so you can process COND_OP dynamically.

    import operator
    from functools import reduce
    
    op_map = {"=": operator.eq, "!=": operator.ne, "<": operator.lt, ">": operator.gt, "<=": operator.le, ">=": operator.ge}
    
    where_list = [op_map[c["COND_OP"]](sql.Column(obj, c["COND_COL"]), c["COND_VAL"]) for c in inputs["CONDS"]]
    
    if where_list:
        stmt.where = reduce(operator.and_, where_list)