from py_jftech import read, write, where, mapper_columns, format_date __COLUMNS__ = { 'rb_id': 'id', 'rb_module': 'module', 'rb_date': 'date', 'rb_risk': 'risk', 'rb_nav': 'nav', 'rb_remarks': 'remarks', } @write def batch_insert(datas): datas = [mapper_columns(x, __COLUMNS__) for x in datas] values = ','.join([f'''({','.join([(f"'{x[j]}'" if j in x and x[j] is not None else 'null') for j in __COLUMNS__.keys() if j != 'rb_id'])})''' for x in datas]) return f'''insert into robo_benchmark({','.join([x for x in __COLUMNS__.keys() if x != 'rb_id'])}) values {values}''' @read(one=True) def get_last_one(module=None, max_date=None, risk=None, re: bool = None): sqls = [] if max_date: sqls.append(f"rb_date <= '{format_date(max_date)}'") return f''' select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_benchmark {where(*sqls, rb_module=module, rb_risk=risk, v_rb_re=re)} order by rb_date desc limit 1 ''' @read def get_list(max_date=None, min_date=None, module=None, risk=None, re: bool = None): sqls = [] if max_date: sqls.append(f"rb_date <= '{format_date(max_date)}'") if min_date: sqls.append(f"rb_date >= '{format_date(min_date)}'") return f''' select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_benchmark {where(*sqls, rb_module=module, rb_risk=risk, v_rb_re=re)} order by rb_risk, rb_date '''