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!
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)