Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use sql method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

app.py

Source:app.py

...17    return 'Welcome to MyCode!'18@app.route('/login', methods=['GET', 'POST'])19def do_login():20    res = make_response()21    dosql = open_sql('mycode')22    if request.method == 'GET':23        try:24            username = request.args.get('username')25            password = request.args.get('password')26            results = do_sql(dosql, f"call doLoginProcedure('{username}','{password}')")27            close_sql(dosql)28            result = simplejson.loads(results)[0]29            if result.get('code') is not None:30                if result.get('code') == '0404':31                    res.status_code = '1404'32                if result.get('code') == '0403':33                    res.status_code = '1403'34            res.data = results35        except:36            res.status_code = '1500'37        return res38    if request.method == 'POST':39        res.status_code = '1500'40        return res41@app.route('/product/')42def product_info():43    dosql = open_sql('mysales')44    if request.method == 'GET':45        productid = request.args.get('productid')46        if (productid == None):47            sql = 'select * from products'48        else:49            sql = f'select * from products where productid = {productid}'50        results = do_sql(dosql, sql)51        close_sql(dosql)52        return results53    elif request.method == 'POST':54        return 'æä¸æ¯ææ°æ®åºæ¥è¯¢ä»¥å¤çæä½'55@app.route('/user/message')56def user_message_info():57    if request.method == 'GET':58        dosql = open_sql('mycode')59        useridList = request.args.get('useridList').split(',')60        myid = request.args.get('userid')61        userMessage = []62        for userid in useridList:63            sql = f"call getUserMessageInfo('{userid}','{myid}');"64            result = simplejson.loads(do_sql(dosql, sql))[0]65            userMessage.append(result)66        close_sql(dosql)67        return simplejson.dumps(userMessage)68@app.route('/user/message/detail')69def user_message_detail():70    if request.method == 'GET':71        dosql = open_sql('mycode')72        userid = request.args.get('userid')73        id = request.args.get('id')74        sql = f"call update_messages('{userid}','{id}');"75        dosql = update_sql(dosql, sql)76        sql = f"call get_before_messages('{userid}','{id}');"77        messages = simplejson.loads(do_sql(dosql, sql))78        messages = messages[::-1]79        sql = f"call get_message_detail('{userid}','{id}');"80        result = simplejson.loads(do_sql(dosql, sql))[0]81        result["messages"] = messages82        close_sql(dosql)83        return simplejson.dumps([result])84@app.route('/problem/type/admin', methods=['GET', 'POST'])85def problem_type_admin():86    dosql = open_sql('mycode')87    if request.method == 'GET':88        sql = 'call select_problemtypes_admin();'89        results = do_sql(dosql, sql)90        close_sql(dosql)91        return results92    elif request.method == 'POST':93        params = request.json.get('params')94        sql = 'call empty_problemtypes();'95        dosql = update_sql(dosql, sql)96        for node in params.get('data'):97            parentid = node.get('parentid')98            if parentid != None:99                sql = f"insert into problemtypes(typeid,text,level,parentid)value('{node.get('typeid')}','{node.get('text')}',{node.get('level')},'{parentid}');"100            else:101                sql = f"insert into problemtypes(typeid,text,level,parentid)value('{node.get('typeid')}','{node.get('text')}',{node.get('level')},null);"102            dosql = update_sql(dosql, sql)103        close_sql(dosql)104        return 'æä¸æ¯ææ°æ®åºæ¥è¯¢ä»¥å¤çæä½'105@app.route('/problem/run', methods=['GET', 'POST'])106def problem_run():107    res = make_response()108    if request.method == 'POST':109        params = request.json.get('params')110        userid = params.get('userid')111        problemid = params.get('problemid')112        type = params.get('typeid')113        userid = params.get('userid')114        rightOutput = params.get('output')115        language = params.get('language')116        submitid = str(random.randint(1111111111, 9999999999))117        func = params.get('func')118        code = params.get('code')119        input = params.get('input')120        if language == 2:121            myoutput, isError = do_python(func, code, input);122        elif language == 3:123            myoutput, isError = do_js(func, code, input, type);124        results = simplejson.loads(simplejson.dumps({}))125        results['data'] = str(myoutput)126        results['isError'] = isError127        res.headers['Content-Type'] = 'application/json; charset=utf-8'128        results['data'] = str(myoutput)129        results['isError'] = isError130        results = simplejson.dumps([results])131        res.data = results132        dosql = open_sql('mycode')133        status = 0134        if str(myoutput) == str(rightOutput):135            status = 1136        else:137            status = 2138        if isError:139            status = 3140        sql = f"call add_code_submit('{problemid}','{submitid}','{userid}',{language},{status});"141        dosql = update_sql(dosql, sql)142        close_sql(dosql)143    return res144@app.route('/problem/runJS', methods=['GET', 'POST'])145def problem_run_js():146    res = make_response()147    if request.method == 'POST':148        params = request.json.get('params')149        dosql = open_sql('mycode')150        sql = f"call get_problem_info('{params.get('problemid')}');"151        problem_info = simplejson.loads(do_sql(dosql, sql))[0]152        input = problem_info.get('input').split('\n')[0]153        output = problem_info.get('output').split('\n')[0]154        code = params.get('code')155        print(code)156        myoutput, isError = do_js(problem_info.get('func'), code, input, problem_info.get('type'));157        results = simplejson.loads(simplejson.dumps({}))158        results['output'] = myoutput159        results['isError'] = isError160        print(results)161        results = simplejson.dumps([results])162        res.data = results163    return res164@app.route('/problem/runPy', methods=['GET', 'POST'])165def problem_run_python():166    res = make_response()167    if request.method == 'POST':168        params = request.json.get('params')169        func = params.get('func')170        input = params.get('input')171        code = params.get('code')172        output, isError = do_python(func, code, input);173        results = simplejson.loads(simplejson.dumps({}))174        results['data'] = output175        results['isError'] = isError176        results = simplejson.dumps([results])177        res.data = results178    return res179@app.route('/problem/runSQL', methods=['GET', 'POST'])180def problem_run_mysql():181    res = make_response()182    if request.method == 'POST':183        params = request.json.get('params')184        code = params.get('code')185        dosql = open_sql('mysales')186        results = do_sql(dosql, code)187        close_sql(dosql)188        res.data = results189    return res190@app.route('/problem/submit/note', methods=['GET', 'POST'])191def problem_submit_note():192    res = make_response()193    if request.method == 'POST':194        params = request.json.get('params')195        sql = f"call problem_submit_note({params.get('submitid')},'{params.get('problemid')}','{params.get('userid')}','{params.get('note')}');"196        dosql = open_sql('mycode')197        dosql = update_sql(dosql, sql)198        close_sql(dosql)199    return res200@app.route('/problems/admin', methods=['GET', 'POST'])201def get_problems_admin():202    res = make_response()203    dosql = open_sql('mycode')204    if request.method == 'GET':205        isparams = 0206        userid = request.args.get('userid')207        if userid == None or userid == '':208            userid = 0209        if True:210            problemname = request.args.get('problemname')211            if problemname == None:212                problemname = 'all'213            type = request.args.get('type')214            if type == None:215                type = '0'216            rank = request.args.get('rank')217            if rank == None:218                rank = 0219            labels = request.args.get('labels')220            if labels == None:221                labels = ''222            status = request.args.get('status')223            if status == None:224                status = 0225            sql = f"call select_problems_admin({userid},1,'{problemname}','{type}',{rank},'{labels}',{status});"226        results = do_sql(dosql, sql)227        close_sql(dosql)228        res.data = results229    return res230@app.route('/problems', methods=['GET', 'POST'])231def get_problems():232    res = make_response()233    dosql = open_sql('mycode')234    if request.method == 'GET':235        isparams = 0236        userid = request.args.get('userid')237        if userid == None or userid == '':238            userid = 0239        if True:240            problemname = request.args.get('problemname')241            if problemname == None:242                problemname = 'all'243            type = request.args.get('type')244            if type == None:245                type = '0'246            rank = request.args.get('rank')247            if rank == None:248                rank = 0249            labels = request.args.get('labels')250            if labels == None:251                labels = ''252            status = request.args.get('status')253            if status == None:254                status = 0255            page = request.args.get('pageNo')256            if page == None:257                page = 1258            sql = f"call select_problems({userid},1,'{problemname}','{type}',{rank},'{labels}',{status},{page});"259        result = simplejson.loads(do_sql(dosql, sql))260        results = simplejson.loads(simplejson.dumps({}))261        # results["results"] = result262        results["total"] = len(result)263        page = int(page) - 1264        count = page * 8265        if len(result) < count + 8:266            result = result[count:]267        else:268            result = result[count:count + 8]269        results["results"] = result270        results = simplejson.dumps([results])271        close_sql(dosql)272        res.data = results273    return res274@app.route('/problem/', methods=['GET', 'POST'])275def get_problem():276    res = make_response()277    dosql = open_sql('mycode')278    if request.method == 'GET':279        problemid = request.args.get('problemid')280        sql = f"call problem_detail(\'{problemid}\')"281        results = do_sql(dosql, sql)282        close_sql(dosql)283        res.data = results284    if request.method == 'POST':285        params = simplejson.loads(request.values.get('params'))286        input = params.get('input')287        if input != None:288            input = input.replace("\'", "\\\'")289        output = params.get('output')290        if output != None:291            output = output.replace("\'", "\\\'")292        if params.get('problemid') == None:293            params["problemid"] = str(random.randint(1111111111, 9999999999))294            sql = f"call add_problem('{params.get('problemid')}','{params.get('title')}','{params.get('msg')}','{input}','{output}',{params.get('rankid')},'{params.get('labels')}','{params.get('typeid')}','{params.get('func')}','{params.get('arguements')}','{params.get('template')}');"295        else:296            sql = f"call edit_problem('{params.get('problemid')}','{params.get('title')}','{params.get('msg')}','{input}','{output}',{params.get('rankid')},'{params.get('labels')}','{params.get('typeid')}','{params.get('func')}','{params.get('arguements')}','{params.get('template')}');"297        print(sql)298        dosql = update_sql(dosql, sql)299        close_sql(dosql)300    return res301@app.route('/problem/submit', methods=['GET', 'POST'])302def problem_submit():303    res = make_response()304    dosql = open_sql('mycode')305    if request.method == 'GET':306        problemid = request.args.get('problemid')307        userid = request.args.get('userid')308        sql = f"call get_code_submit('{problemid}','{userid}')"309        results = do_sql(dosql, sql)310        close_sql(dosql)311        res.data = results312    if request.method == 'POST':313        params = request.json.get('params')314        dosql = open_sql('mycode')315        problemid = params.get('problemid')316        language = params.get('language')317        userid = params.get('userid')318        sql = f"call get_problem_info('{params.get('problemid')}');"319        problem_info = simplejson.loads(do_sql(dosql, sql))[0]320        input = problem_info.get('input').split('\n')321        func = problem_info.get('func')322        type = problem_info.get('typeid')323        if type != 'C':324            output = problem_info.get('output').split('\n')325            code = params.get('code')326            sum = len(input)327            count = 0328            wrong_output = ''329            wrong_input = ''330            except_output = ''331            for index in range(0, len(input)):332                if int(language) == 3:333                    result, isError = do_js(func, code, input[index], type)334                elif int(language) == 2:335                    result, isError = do_py(func, code, input[index])336                    print(result)337                if result == output[index]:338                    count += 1339                else:340                    wrong_input = input[index]341                    except_output = output[index]342                    wrong_output = result343                    break344            if count == sum:345                status = 1346            else:347                status = 2348            codes = params.get('code').replace('\'', '\\\'');349            sql = f"call add_code_submit('{problemid}','{userid}','{codes}',{language},{status})"350            dosql = update_sql(dosql, sql)351            close_sql(dosql)352            result = simplejson.loads(simplejson.dumps({}))353            result["except"] = count354            result["input"] = wrong_input355            result["output"] = except_output356            result["sum"] = sum357            result["result"] = wrong_output358            result["status"] = status359            result = simplejson.dumps([result])360            res.data = result361        else:362            template = problem_info.get('template')363            problem_sql = open_sql('mysales')364            result1 = simplejson.loads(do_sql(problem_sql, template))365            try:366                result2 = simplejson.loads(do_sql(problem_sql, params.get('code')))367                is_equals = is_rs_eauals(result1, result2)368            except Exception as e:369                result2 = str(e).replace('mysales', 'fake_database')370                is_equals = False371            close_sql(problem_sql)372            result = simplejson.loads(simplejson.dumps({}))373            result["except"] = 1 if is_equals else 0374            result["input"] = template375            result["output"] = simplejson.dumps(result1)376            result["sum"] = 1377            try:378                result["result"] = simplejson.dumps(result2)379            except:380                result["result"] = result2381            result["status"] = 1 if is_equals else 2382            result = simplejson.dumps([result])383            res.data = result384            sql = f"call add_code_submit('{problemid}','{userid}','{params.get('code')}',{language},{1 if is_equals else 2})"385            dosql = update_sql(dosql, sql)386            close_sql(dosql)387            close_sql(dosql)388    return res389@app.route('/plan/problem/submit', methods=['GET', 'POST'])390def plan_problem_submit():391    res = make_response()392    dosql = open_sql('mycode')393    if request.method == 'GET':394        problemid = request.args.get('problemid')395        userid = request.args.get('userid')396        sql = f"call get_code_submit('{problemid}','{userid}')"397        results = do_sql(dosql, sql)398        close_sql(dosql)399        res.data = results400    if request.method == 'POST':401        params = request.json.get('params')402        dosql = open_sql('mycode')403        problemid = params.get('problemid')404        planid = params.get('planid')405        language = params.get('language')406        userid = params.get('userid')407        sql = f"call get_problem_info('{params.get('problemid')}');"408        problem_info = simplejson.loads(do_sql(dosql, sql))[0]409        input = problem_info.get('input').split('\n')410        func = problem_info.get('func')411        type = problem_info.get('type')412        output = problem_info.get('output').split('\n')413        code = params.get('code')414        sum = len(input)415        count = 0416        wrong_output = ''417        wrong_input = ''418        except_output = ''419        for index in range(0, len(input)):420            result, isError = do_js(func, code, input[index], type)421            if result == output[index]:422                count += 1423            else:424                wrong_input = input[index]425                except_output = output[index]426                wrong_output = result427                break428        if count == sum:429            status = 1430        else:431            status = 2432        codes = params.get('code').replace('\'', '\\\'');433        sql = f"call add_code_submit('{problemid}','{userid}','{codes}',{language},{status})"434        dosql = update_sql(dosql, sql)435        # æ¤å¤è¿è¡æ¯å¦å®æ436        if status == 1:437            sql = f"call add_plan_user_problem('{planid}','{userid}','{problemid}');"438            print(sql)439            dosql = update_sql(dosql, sql)440        close_sql(dosql)441        result = simplejson.loads(simplejson.dumps({}))442        result["except"] = count443        result["input"] = wrong_input444        result["output"] = except_output445        result["sum"] = sum446        result["result"] = wrong_output447        result["status"] = status448        print(result)449        result = simplejson.dumps([result])450        res.data = result451    return res452@app.route('/problem/labels', methods=['GET', 'POST'])453def get_problem_labels():454    res = make_response()455    dosql = open_sql('mycode')456    if request.method == 'GET':457        sql = f"call get_problem_labels();"458        results = do_sql(dosql, sql)459        close_sql(dosql)460        res.data = results461    return res462@app.route('/learn/plan', methods=['GET', 'POST'])463def get_learn_plan():464    res = make_response()465    dosql = open_sql('mycode')466    if request.method == 'GET':467        sql = f"call get_learn_plan();"468        results = do_sql(dosql, sql)469        res.data = results470    if request.method == 'POST':471        params = request.json.get('params')472        planid = str(random.randint(1111111111, 9999999999))473        planname = params.get('planname')474        msg = params.get('msg')475        labels = params.get('labels')476        partList = params.get('partList')477        problemList = params.get('problemList')478        sql = f"call add_learn_plan('{planid}','{planname}','{msg}','{labels}');"479        dosql = update_sql(dosql, sql)480        for part in partList:481            sql = f"call add_plan_part('{planid}',{part.get('partid')},'{part.get('partname')}','{part.get('msg')}');"482            print(sql)483            dosql = update_sql(dosql, sql)484        for problem in problemList:485            sql = f"call add_plan_problem('{planid}','{problem.get('problemid')}',{problem.get('part')},{problem.get('points')},{problem.get('needpoints')});"486            dosql = update_sql(dosql, sql)487    close_sql(dosql)488    return res489@app.route('/learn/plan/mine', methods=['GET', 'POST'])490def get_my_plan():491    res = make_response()492    dosql = open_sql('mycode')493    if request.method == 'GET':494        userid = request.args.get('userid')495        sql = f"call get_user_plan('{userid}');"496        results = do_sql(dosql, sql)497        res.data = results498    if request.method == 'POST':499        params = request.json.get('params')500        userid = params.get('userid')501        planid = params.get('planid')502        sql = f"call add_user_plan('{userid}','{planid}');"503        print(sql)504        results = do_sql(dosql, sql)505        res.data = results506    close_sql(dosql)507    return res508@app.route('/learn/plan/problems', methods=['GET', 'POST'])509def get_plan_problems():510    res = make_response()511    dosql = open_sql('mycode')512    if request.method == 'GET':513        sql = f"call get_plan_problems_admin();"514        results = do_sql(dosql, sql)515        close_sql(dosql)516        res.data = results517    return res518@app.route('/learn/plan/detail', methods=['GET', 'POST'])519def get_plan_detail():520    res = make_response()521    dosql = open_sql('mycode')522    if request.method == 'GET':523        results = simplejson.loads(simplejson.dumps({}))524        planid = request.args.get('planid')525        userid = request.args.get('userid')526        sql = f"call get_plan_detail('{planid}','{userid}');"527        detail = simplejson.loads(do_sql(dosql, sql))[0]528        labelid_list = detail.get('labels').split(',')529        labels = ''530        for labelid in labelid_list:531            sql = f"call get_problem_label('{labelid}');"532            labels += (',' + simplejson.loads(do_sql(dosql, sql))[0].get('text'))533        if labels != '':534            labels = labels[1:]535        detail["labels"] = labels536        results["detail"] = detail537        sql = f"call get_plan_problems('{planid}','{userid}');"538        results["problemList"] = simplejson.loads(do_sql(dosql, sql))539        sql = f"call get_plan_parts('{planid}');"540        results["partList"] = simplejson.loads(do_sql(dosql, sql))541        results = simplejson.dumps([results])542        res.data = results543    close_sql(dosql)544    return res545@app.route('/problems/total', methods=['GET', 'POST'])546def get_problem_total():547    res = make_response()548    dosql = open_sql('mycode')549    if request.method == 'GET':550        sql = f"call get_problem_labels();"551        results = do_sql(dosql, sql)552        close_sql(dosql)553        res.data = results554    return res555@app.route('/circle/join', methods=['GET', 'POST'])556def circle_join():557    res = make_response()558    dosql = open_sql('mycode')559    if request.method == 'POST':560        params = request.json.get('params')561        circleid = params.get('id')562        userid = params.get('userid')563        sql = f"call user_circles('{circleid}','{userid}');"564        dosql = update_sql(dosql, sql)565        close_sql(dosql)566    return res567@app.route('/problem/delete', methods=['GET', 'POST'])568def delete_problem():569    res = make_response()570    dosql = open_sql('mycode')571    if request.method == 'GET':572        problemid = request.args.get('problemid')573        sql = f"call delete_problem({problemid});"574        dosql = update_sql(dosql, sql)575        close_sql(dosql)576    return res577@app.route('/problem/types', methods=['GET', 'POST'])578def get_problem_types():579    res = make_response()580    dosql = open_sql('mycode')581    if request.method == 'GET':582        sql = f"call get_problem_types()"583        results = do_sql(dosql, sql)584        close_sql(dosql)585        res.data = results586    return res587@app.route('/circle', methods=['GET', 'POST'])588def get_circle():589    res = make_response()590    dosql = open_sql('mycode')591    if request.method == 'GET':592        id = request.args.get('id')593        userid = request.args.get('userid')594        if id == None:595            id = '0'596        if userid == None:597            userid = '0'598        sql = f"call get_circles('{id}','{userid}');"599        results = do_sql(dosql, sql)600        close_sql(dosql)601        res.data = results602    if request.method == 'POST':603        params = request.json.get('params')604        print(params)605        userid = params.get('userid')606        msg = params.get('msg')607        circlename = params.get('circlename')608        ispublic = params.get('ispublic')609        parentid = params.get('parentid')610        circleid = str(random.randint(1111111111, 9999999999))611        if parentid == 'root':612            sql = f"call add_circles('{circleid}','{circlename}','{msg}',null,1,'{userid}',0,{ispublic});"613        else:614            sql = f"call add_circles('{circleid}','{circlename}','{msg}','{parentid}',0,'{userid}',0,{ispublic});"615        print(sql)616        dosql = update_sql(dosql, sql)617        close_sql(dosql)618    return res619@app.route('/forums', methods=['GET', 'POST'])620def get_forums():621    res = make_response()622    dosql = open_sql('mycode')623    if request.method == 'GET':624        circleid = request.args.get('circleid')625        sql = f"call get_forums('{circleid}');"626        results = simplejson.loads(do_sql(dosql, sql))627        if len(results) != 0:628            for result in results:629                result_labels = result.get('labels').split(',')630                labels = []631                for label in result_labels:632                    if label != '':633                        text = simplejson.loads(do_sql(dosql, f"call get_circle_label('{circleid}','{label}');"))634                        text = text[0].get('text')635                        labels.append(text)636                result['labels'] = ','.join(labels)637        res.data = simplejson.dumps(results)638        print(results)639        close_sql(dosql)640    return res641@app.route('/forum', methods=['GET', 'POST'])642def get_forum():643    res = make_response()644    dosql = open_sql('mycode')645    if request.method == 'GET':646        forumid = request.args.get('forumid')647        sql = f"call get_forum('{forumid}');"648        results = do_sql(dosql, sql)649        close_sql(dosql)650        res.data = results651    return res652@app.route('/forum/comment', methods=['GET', 'POST'])653def get_forum_comment():654    res = make_response()655    dosql = open_sql('mycode')656    if request.method == 'GET':657        forumid = request.args.get('forumid')658        sql = f"call get_forum_comment('{forumid}');"659        results = do_sql(dosql, sql)660        close_sql(dosql)661        res.data = results662    return res663@app.route('/forum/comments', methods=['GET', 'POST'])664def get_forum_comments():665    res = make_response()666    dosql = open_sql('mycode')667    if request.method == 'GET':668        commentid = request.args.get('commentid')669        sql = f"call get_forum_comment_comment('{commentid}');"670        results = do_sql(dosql, sql)671        close_sql(dosql)672        res.data = results673    return res674@app.route('/problems/types', methods=['GET', 'POST'])675def problems_types():676    res = make_response()677    dosql = open_sql('mycode')678    if request.method == 'GET':679        typeid = request.args.get('typeid')680        if (typeid == None):681            typeid = '0'682        sql = f"call select_problemtypes('{typeid}')"683        results = do_sql(dosql, sql)684        close_sql(dosql)685        res.data = results686    return res687@app.route('/problems/labels', methods=['GET', 'POST'])688def problems_labels():689    res = make_response()690    dosql = open_sql('mycode')691    if request.method == 'GET':692        sql = f"select * from problemlabels;"693        results = do_sql(dosql, sql)694        close_sql(dosql)695        res.data = results696    return res697@app.route('/reply/comment/', methods=['GET', 'POST'])698def reply_comments():699    dosql = open_sql('mycode')700    if request.method == 'GET':701        replyid = request.args.get('replyid')702        if (replyid == None):703            problemid = request.args.get('problemid')704            sql = f"call problem_replys('{problemid}');"705        else:706            sql = f"call reply_comments('{replyid}');"707        results = do_sql(dosql, sql)708        close_sql(dosql)709        return results710    if request.method == 'POST':711        res = make_response()712        params = request.json.get('params')713        replyid = params.get('replyid')714        replyuserid = params.get('replyuserid')715        userid = params.get('userid')716        commentid = str(random.randint(1111111111, 9999999999))717        content = params.get('content')718        sql = f"call add_reply_comment('{commentid}','{replyid}','{userid}','{replyuserid}','{content}');"719        update_sql(dosql, sql)720        close_sql(dosql)721        return res722@app.route('/circle/labels', methods=['GET', 'POST'])723def circle_labels():724    dosql = open_sql('mycode')725    if request.method == 'GET':726        res = make_response()727        circleid = request.args.get('circleid')728        sql = f"call get_circle_labels('{circleid}');"729        print(sql)730        results = do_sql(dosql, sql)731        close_sql(dosql)732        res.data = results733        return res734@app.route('/circle/label/forum', methods=['GET', 'POST'])735def label_forums():736    dosql = open_sql('mycode')737    if request.method == 'GET':738        res = make_response()739        circleid = request.args.get('circleid')740        labelid = request.args.get('labelid')741        sql = f"call get_labels_forums('{circleid}','{labelid}');"742        results = do_sql(dosql, sql)743        close_sql(dosql)744        res.data = results745        return res746@app.route('/forumorcomment', methods=['GET', 'POST'])747def forum_comment():748    dosql = open_sql('mycode')749    if request.method == 'GET':750        res = make_response()751        id = request.args.get('id')752        type = request.args.get('type')753        if type == 'forum':754            sql = f"call get_forum_info('{id}')"755        if type == 'comment':756            sql = f"call get_comment_info('{id}')"757        results = do_sql(dosql, sql)758        close_sql(dosql)759        res.data = results760        return res761@app.route('/problem/solution/', methods=['GET', 'POST'])762def problem_solution():763    dosql = open_sql('mycode')764    if request.method == 'GET':765        res = make_response()766        solutionid = request.args.get('solutionid')767        sql = f"call get_solution('{solutionid}');"768        results = simplejson.loads(do_sql(dosql, sql))769        result = results[0]770        labels = result['labels']771        try:772            labels = labels['labels'].split(';')773            labels = str(get_label(dosql, labels))774            result['labels'] = labels775        except:776            result['labels'] = ''777        results = [result]778        res.data = simplejson.dumps(results)779        return res780@app.route('/circle/forum', methods=['GET', 'POST'])781def circle_forum_nopublic():782    res = make_response()783    dosql = open_sql('mycode')784    if request.method == 'POST':785        params = request.values786        forumid = str(random.randint(1111111111, 9999999999))787        circleid = params.get('circleid')788        userid = params.get('userid')789        content = params.get('content')790        title = params.get('title')791        sql = f"call add_forum_nopublic('{circleid}','{forumid}','{userid}','{content}','{title}');"792        dosql = update_sql(dosql, sql)793        close_sql(dosql)794    return res795@app.route('/circle/forum/admin', methods=['GET', 'POST'])796def circle_forum_admin():797    res = make_response()798    dosql = open_sql('mycode')799    if request.method == 'GET':800        sql = "call get_forum_nopublic();"801        results = do_sql(dosql, sql)802        res.data = results803    close_sql(dosql)804    return res805@app.route('/forum/admin/unpass', methods=['GET', 'POST'])806def circle_forum_admin_un():807    res = make_response()808    dosql = open_sql('mycode')809    if request.method == 'POST':810        id = request.json.get('params').get('forumid')811        sql = f"delete from nopublicforums where forumid = '{id}';"812        print(sql)813        dosql = update_sql(dosql, sql)814    close_sql(dosql)815    return res816@app.route('/circle/forum/public', methods=['GET', 'POST'])817def circle_forum():818    res = make_response()819    dosql = open_sql('mycode')820    if request.method == 'POST':821        params = request.json.get('params')822        forumid = params.get('id')823        sql = f"select * from nopublicforums where forumid = '{forumid}';"824        forum_detail = simplejson.loads(do_sql(dosql, sql))[0]825        circleid = forum_detail.get('circleid')826        userid = forum_detail.get('userid')827        content = forum_detail.get('content')828        title = forum_detail.get('title')829        labels = content_to_labels(title, content)830        circle_labels = simplejson.loads(do_sql(dosql, f"call get_circle_labels('{circleid}');"))831        labels = concat_label(circle_labels, labels)832        result_labels = []833        for label in labels:834            if type(label) == str:835                labelid = str(random.randint(1111111111, 9999999999))836                dosql = update_sql(dosql, f"call add_circle_label('{circleid}','{labelid}','{label}');")837                result_labels.append(labelid)838            elif type(label) == dict:839                result_labels.append(label.get('id'))840        sql = f"call add_forum('{circleid}','{forumid}','{userid}','{content}','{title}','{','.join(result_labels)}');"841        dosql = update_sql(dosql, sql)842    return res843@app.route('/problem/solutions/', methods=['GET', 'POST'])844def problem_solutions():845    dosql = open_sql('mycode')846    if request.method == 'GET':847        res = make_response()848        problemid = request.args.get('problemid')849        sql = f"call get_solutions('{problemid}');"850        results = simplejson.loads(do_sql(dosql, sql))851        for result in results:852            labels = result['labels'].split(';')853            try:854                labels = str(get_label(dosql, labels))855                result['labels'] = labels856            except:857                result['labels'] = ''858        res.data = simplejson.dumps(results)859        close_sql(dosql)860        return res861    if request.method == 'POST':862        res = make_response()863        params = request.values864        dosql = open_sql('mycode')865        solutionid = str(random.randint(1111111111, 9999999999))866        problemid = params.get('problemid')867        userid = params.get('problemid')868        isofficial = params.get('isofficial')869        language = params.get('language')870        title = params.get('title')871        if title == None:872            title = 'ææ æ é¢'873        content = params.get('content')874        labels = params.get('labels')875        sql = f"call add_problem_solution('{solutionid}','{problemid}','{isofficial}','{userid}','{title}','{language}','{content}','','{labels}')"876        dosql = update_sql(dosql, sql)877        close_sql(dosql)878        return res879@app.route('/solution/label/', methods=['GET', 'POST'])880def solution_label():881    dosql = open_sql('mycode')882    if request.method == 'GET':883        res = make_response()884        sql = f"call get_solution_labels();"885        results = do_sql(dosql, sql)886        close_sql(dosql)887        res.data = results888    return res889@app.route('/label/solution/', methods=['GET', 'POST'])890def label_solution():891    dosql = open_sql('mycode')892    if request.method == 'GET':893        res = make_response()894        language = request.args.get('language')895        problemid = request.args.get('problemid')896        if (language == None):897            language = 0898        sql = f"call get_label_solutions('{problemid}','{language}',0);"899        results = do_sql(dosql, sql)900        res.data = results901        labels = request.args.get('labels')902        if (labels != ''):903            labels_id = []904            results = simplejson.loads(results)905            for item in results:906                labels_id.append(item['solutionid'])907            labels = labels[1:]908            label_list = labels.split(',')909            for label in label_list:910                sql = f"call get_label_solutions('{problemid}','{language}',{label});"911                result = simplejson.loads(do_sql(dosql, sql))912                label_id = []913                for item in result:914                    label_id.append(item['solutionid'])915                labels_id = list(set(labels_id).intersection(set(label_id)))916            new_results = []917            for result in results:918                if (result['solutionid'] in labels_id):919                    labels = result['labels'].split(';')920                    labels = str(get_label(dosql, labels))921                    result['labels'] = labels922                    new_results.append(result)923            new_results = simplejson.dumps(new_results)924            res.data = new_results925        close_sql(dosql)926    return res927if __name__ == '__main__':...

test_sql.py

Source:test_sql.py

1#!/usr/bin/env python2# test_sql.py - tests for the psycopg2.sql module3#4# Copyright (C) 2016 Daniele Varrazzo  <daniele.varrazzo@gmail.com>5#6# psycopg2 is free software: you can redistribute it and/or modify it7# under the terms of the GNU Lesser General Public License as published8# by the Free Software Foundation, either version 3 of the License, or9# (at your option) any later version.10#11# In addition, as a special exception, the copyright holders give12# permission to link this program with the OpenSSL library (or with13# modified versions of OpenSSL that use the same license as OpenSSL),14# and distribute linked combinations including the two.15#16# You must obey the GNU Lesser General Public License in all respects for17# all of the code used other than OpenSSL.18#19# psycopg2 is distributed in the hope that it will be useful, but WITHOUT20# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or21# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public22# License for more details.23import datetime as dt24from io import StringIO25from .testutils import (unittest, ConnectingTestCase,26    skip_before_postgres, skip_before_python, skip_copy_if_green)27import psycopg228from psycopg2 import sql29class SqlFormatTests(ConnectingTestCase):30    @skip_before_python(2, 7)31    def test_pos(self):32        s = sql.SQL("select {} from {}").format(33            sql.Identifier('field'), sql.Identifier('table'))34        s1 = s.as_string(self.conn)35        self.assertTrue(isinstance(s1, str))36        self.assertEqual(s1, 'select "field" from "table"')37    def test_pos_spec(self):38        s = sql.SQL("select {0} from {1}").format(39            sql.Identifier('field'), sql.Identifier('table'))40        s1 = s.as_string(self.conn)41        self.assertTrue(isinstance(s1, str))42        self.assertEqual(s1, 'select "field" from "table"')43        s = sql.SQL("select {1} from {0}").format(44            sql.Identifier('table'), sql.Identifier('field'))45        s1 = s.as_string(self.conn)46        self.assertTrue(isinstance(s1, str))47        self.assertEqual(s1, 'select "field" from "table"')48    def test_dict(self):49        s = sql.SQL("select {f} from {t}").format(50            f=sql.Identifier('field'), t=sql.Identifier('table'))51        s1 = s.as_string(self.conn)52        self.assertTrue(isinstance(s1, str))53        self.assertEqual(s1, 'select "field" from "table"')54    def test_unicode(self):55        s = sql.SQL("select {0} from {1}").format(56            sql.Identifier('field'), sql.Identifier('table'))57        s1 = s.as_string(self.conn)58        self.assertTrue(isinstance(s1, str))59        self.assertEqual(s1, 'select "field" from "table"')60    def test_compose_literal(self):61        s = sql.SQL("select {0};").format(sql.Literal(dt.date(2016, 12, 31)))62        s1 = s.as_string(self.conn)63        self.assertEqual(s1, "select '2016-12-31'::date;")64    def test_compose_empty(self):65        s = sql.SQL("select foo;").format()66        s1 = s.as_string(self.conn)67        self.assertEqual(s1, "select foo;")68    def test_percent_escape(self):69        s = sql.SQL("42 % {0}").format(sql.Literal(7))70        s1 = s.as_string(self.conn)71        self.assertEqual(s1, "42 % 7")72    def test_braces_escape(self):73        s = sql.SQL("{{{0}}}").format(sql.Literal(7))74        self.assertEqual(s.as_string(self.conn), "{7}")75        s = sql.SQL("{{1,{0}}}").format(sql.Literal(7))76        self.assertEqual(s.as_string(self.conn), "{1,7}")77    def test_compose_badnargs(self):78        self.assertRaises(IndexError, sql.SQL("select {0};").format)79    @skip_before_python(2, 7)80    def test_compose_badnargs_auto(self):81        self.assertRaises(IndexError, sql.SQL("select {};").format)82        self.assertRaises(ValueError, sql.SQL("select {} {1};").format, 10, 20)83        self.assertRaises(ValueError, sql.SQL("select {0} {};").format, 10, 20)84    def test_compose_bad_args_type(self):85        self.assertRaises(IndexError, sql.SQL("select {0};").format, a=10)86        self.assertRaises(KeyError, sql.SQL("select {x};").format, 10)87    def test_must_be_composable(self):88        self.assertRaises(TypeError, sql.SQL("select {0};").format, 'foo')89        self.assertRaises(TypeError, sql.SQL("select {0};").format, 10)90    def test_no_modifiers(self):91        self.assertRaises(ValueError, sql.SQL("select {a!r};").format, a=10)92        self.assertRaises(ValueError, sql.SQL("select {a:<};").format, a=10)93    def test_must_be_adaptable(self):94        class Foo(object):95            pass96        self.assertRaises(psycopg2.ProgrammingError,97            sql.SQL("select {0};").format(sql.Literal(Foo())).as_string, self.conn)98    def test_execute(self):99        cur = self.conn.cursor()100        cur.execute("""101            create table test_compose (102                id serial primary key,103                foo text, bar text, "ba'z" text)104            """)105        cur.execute(106            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(107                sql.Identifier('test_compose'),108                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),109                (sql.Placeholder() * 3).join(', ')),110            (10, 'a', 'b', 'c'))111        cur.execute("select * from test_compose")112        self.assertEqual(cur.fetchall(), [(10, 'a', 'b', 'c')])113    def test_executemany(self):114        cur = self.conn.cursor()115        cur.execute("""116            create table test_compose (117                id serial primary key,118                foo text, bar text, "ba'z" text)119            """)120        cur.executemany(121            sql.SQL("insert into {0} (id, {1}) values (%s, {2})").format(122                sql.Identifier('test_compose'),123                sql.SQL(', ').join(map(sql.Identifier, ['foo', 'bar', "ba'z"])),124                (sql.Placeholder() * 3).join(', ')),125            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])126        cur.execute("select * from test_compose")127        self.assertEqual(cur.fetchall(),128            [(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])129    @skip_copy_if_green130    @skip_before_postgres(8, 2)131    def test_copy(self):132        cur = self.conn.cursor()133        cur.execute("""134            create table test_compose (135                id serial primary key,136                foo text, bar text, "ba'z" text)137            """)138        s = StringIO("10\ta\tb\tc\n20\td\te\tf\n")139        cur.copy_expert(140            sql.SQL("copy {t} (id, foo, bar, {f}) from stdin").format(141                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s)142        s1 = StringIO()143        cur.copy_expert(144            sql.SQL("copy (select {f} from {t} order by id) to stdout").format(145                t=sql.Identifier("test_compose"), f=sql.Identifier("ba'z")), s1)146        s1.seek(0)147        self.assertEqual(s1.read(), 'c\nf\n')148class IdentifierTests(ConnectingTestCase):149    def test_class(self):150        self.assertTrue(issubclass(sql.Identifier, sql.Composable))151    def test_init(self):152        self.assertTrue(isinstance(sql.Identifier('foo'), sql.Identifier))153        self.assertTrue(isinstance(sql.Identifier('foo'), sql.Identifier))154        self.assertRaises(TypeError, sql.Identifier, 10)155        self.assertRaises(TypeError, sql.Identifier, dt.date(2016, 12, 31))156    def test_string(self):157        self.assertEqual(sql.Identifier('foo').string, 'foo')158    def test_repr(self):159        obj = sql.Identifier("fo'o")160        self.assertEqual(repr(obj), 'Identifier("fo\'o")')161        self.assertEqual(repr(obj), str(obj))162    def test_eq(self):163        self.assertTrue(sql.Identifier('foo') == sql.Identifier('foo'))164        self.assertTrue(sql.Identifier('foo') != sql.Identifier('bar'))165        self.assertTrue(sql.Identifier('foo') != 'foo')166        self.assertTrue(sql.Identifier('foo') != sql.SQL('foo'))167    def test_as_str(self):168        self.assertEqual(sql.Identifier('foo').as_string(self.conn), '"foo"')169        self.assertEqual(sql.Identifier("fo'o").as_string(self.conn), '"fo\'o"')170    def test_join(self):171        self.assertTrue(not hasattr(sql.Identifier('foo'), 'join'))172class LiteralTests(ConnectingTestCase):173    def test_class(self):174        self.assertTrue(issubclass(sql.Literal, sql.Composable))175    def test_init(self):176        self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))177        self.assertTrue(isinstance(sql.Literal('foo'), sql.Literal))178        self.assertTrue(isinstance(sql.Literal(b'foo'), sql.Literal))179        self.assertTrue(isinstance(sql.Literal(42), sql.Literal))180        self.assertTrue(isinstance(181            sql.Literal(dt.date(2016, 12, 31)), sql.Literal))182    def test_wrapped(self):183        self.assertEqual(sql.Literal('foo').wrapped, 'foo')184    def test_repr(self):185        self.assertEqual(repr(sql.Literal("foo")), "Literal('foo')")186        self.assertEqual(str(sql.Literal("foo")), "Literal('foo')")187        self.assertQuotedEqual(sql.Literal("foo").as_string(self.conn), "'foo'")188        self.assertEqual(sql.Literal(42).as_string(self.conn), "42")189        self.assertEqual(190            sql.Literal(dt.date(2017, 1, 1)).as_string(self.conn),191            "'2017-01-01'::date")192    def test_eq(self):193        self.assertTrue(sql.Literal('foo') == sql.Literal('foo'))194        self.assertTrue(sql.Literal('foo') != sql.Literal('bar'))195        self.assertTrue(sql.Literal('foo') != 'foo')196        self.assertTrue(sql.Literal('foo') != sql.SQL('foo'))197    def test_must_be_adaptable(self):198        class Foo(object):199            pass200        self.assertRaises(psycopg2.ProgrammingError,201            sql.Literal(Foo()).as_string, self.conn)202class SQLTests(ConnectingTestCase):203    def test_class(self):204        self.assertTrue(issubclass(sql.SQL, sql.Composable))205    def test_init(self):206        self.assertTrue(isinstance(sql.SQL('foo'), sql.SQL))207        self.assertTrue(isinstance(sql.SQL('foo'), sql.SQL))208        self.assertRaises(TypeError, sql.SQL, 10)209        self.assertRaises(TypeError, sql.SQL, dt.date(2016, 12, 31))210    def test_string(self):211        self.assertEqual(sql.SQL('foo').string, 'foo')212    def test_repr(self):213        self.assertEqual(repr(sql.SQL("foo")), "SQL('foo')")214        self.assertEqual(str(sql.SQL("foo")), "SQL('foo')")215        self.assertEqual(sql.SQL("foo").as_string(self.conn), "foo")216    def test_eq(self):217        self.assertTrue(sql.SQL('foo') == sql.SQL('foo'))218        self.assertTrue(sql.SQL('foo') != sql.SQL('bar'))219        self.assertTrue(sql.SQL('foo') != 'foo')220        self.assertTrue(sql.SQL('foo') != sql.Literal('foo'))221    def test_sum(self):222        obj = sql.SQL("foo") + sql.SQL("bar")223        self.assertTrue(isinstance(obj, sql.Composed))224        self.assertEqual(obj.as_string(self.conn), "foobar")225    def test_sum_inplace(self):226        obj = sql.SQL("foo")227        obj += sql.SQL("bar")228        self.assertTrue(isinstance(obj, sql.Composed))229        self.assertEqual(obj.as_string(self.conn), "foobar")230    def test_multiply(self):231        obj = sql.SQL("foo") * 3232        self.assertTrue(isinstance(obj, sql.Composed))233        self.assertEqual(obj.as_string(self.conn), "foofoofoo")234    def test_join(self):235        obj = sql.SQL(", ").join(236            [sql.Identifier('foo'), sql.SQL('bar'), sql.Literal(42)])237        self.assertTrue(isinstance(obj, sql.Composed))238        self.assertEqual(obj.as_string(self.conn), '"foo", bar, 42')239        obj = sql.SQL(", ").join(240            sql.Composed([sql.Identifier('foo'), sql.SQL('bar'), sql.Literal(42)]))241        self.assertTrue(isinstance(obj, sql.Composed))242        self.assertEqual(obj.as_string(self.conn), '"foo", bar, 42')243        obj = sql.SQL(", ").join([])244        self.assertEqual(obj, sql.Composed([]))245class ComposedTest(ConnectingTestCase):246    def test_class(self):247        self.assertTrue(issubclass(sql.Composed, sql.Composable))248    def test_repr(self):249        obj = sql.Composed([sql.Literal("foo"), sql.Identifier("b'ar")])250        self.assertEqual(repr(obj),251            """Composed([Literal('foo'), Identifier("b'ar")])""")252        self.assertEqual(str(obj), repr(obj))253    def test_seq(self):254        l = [sql.SQL('foo'), sql.Literal('bar'), sql.Identifier('baz')]255        self.assertEqual(sql.Composed(l).seq, l)256    def test_eq(self):257        l = [sql.Literal("foo"), sql.Identifier("b'ar")]258        l2 = [sql.Literal("foo"), sql.Literal("b'ar")]259        self.assertTrue(sql.Composed(l) == sql.Composed(list(l)))260        self.assertTrue(sql.Composed(l) != l)261        self.assertTrue(sql.Composed(l) != sql.Composed(l2))262    def test_join(self):263        obj = sql.Composed([sql.Literal("foo"), sql.Identifier("b'ar")])264        obj = obj.join(", ")265        self.assertTrue(isinstance(obj, sql.Composed))266        self.assertQuotedEqual(obj.as_string(self.conn), "'foo', \"b'ar\"")267    def test_sum(self):268        obj = sql.Composed([sql.SQL("foo ")])269        obj = obj + sql.Literal("bar")270        self.assertTrue(isinstance(obj, sql.Composed))271        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")272    def test_sum_inplace(self):273        obj = sql.Composed([sql.SQL("foo ")])274        obj += sql.Literal("bar")275        self.assertTrue(isinstance(obj, sql.Composed))276        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")277        obj = sql.Composed([sql.SQL("foo ")])278        obj += sql.Composed([sql.Literal("bar")])279        self.assertTrue(isinstance(obj, sql.Composed))280        self.assertQuotedEqual(obj.as_string(self.conn), "foo 'bar'")281    def test_iter(self):282        obj = sql.Composed([sql.SQL("foo"), sql.SQL('bar')])283        it = iter(obj)284        i = next(it)285        self.assertEqual(i, sql.SQL('foo'))286        i = next(it)287        self.assertEqual(i, sql.SQL('bar'))288        self.assertRaises(StopIteration, it.__next__)289class PlaceholderTest(ConnectingTestCase):290    def test_class(self):291        self.assertTrue(issubclass(sql.Placeholder, sql.Composable))292    def test_name(self):293        self.assertEqual(sql.Placeholder().name, None)294        self.assertEqual(sql.Placeholder('foo').name, 'foo')295    def test_repr(self):296        self.assertTrue(str(sql.Placeholder()), 'Placeholder()')297        self.assertTrue(repr(sql.Placeholder()), 'Placeholder()')298        self.assertTrue(sql.Placeholder().as_string(self.conn), '%s')299    def test_repr_name(self):300        self.assertTrue(str(sql.Placeholder('foo')), "Placeholder('foo')")301        self.assertTrue(repr(sql.Placeholder('foo')), "Placeholder('foo')")302        self.assertTrue(sql.Placeholder('foo').as_string(self.conn), '%(foo)s')303    def test_bad_name(self):304        self.assertRaises(ValueError, sql.Placeholder, ')')305    def test_eq(self):306        self.assertTrue(sql.Placeholder('foo') == sql.Placeholder('foo'))307        self.assertTrue(sql.Placeholder('foo') != sql.Placeholder('bar'))308        self.assertTrue(sql.Placeholder('foo') != 'foo')309        self.assertTrue(sql.Placeholder() == sql.Placeholder())310        self.assertTrue(sql.Placeholder('foo') != sql.Placeholder())311        self.assertTrue(sql.Placeholder('foo') != sql.Literal('foo'))312class ValuesTest(ConnectingTestCase):313    def test_null(self):314        self.assertTrue(isinstance(sql.NULL, sql.SQL))315        self.assertEqual(sql.NULL.as_string(self.conn), "NULL")316    def test_default(self):317        self.assertTrue(isinstance(sql.DEFAULT, sql.SQL))318        self.assertEqual(sql.DEFAULT.as_string(self.conn), "DEFAULT")319def test_suite():320    return unittest.TestLoader().loadTestsFromName(__name__)321if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.