执行一条sql :param sql: 需要执行的sql :param params: sql参数 :return: 执行结果
(sql: str, params)
| 12 | |
| 13 | |
| 14 | def sql_execute(sql: str, params): |
| 15 | """ |
| 16 | 执行一条sql |
| 17 | :param sql: 需要执行的sql |
| 18 | :param params: sql参数 |
| 19 | :return: 执行结果 |
| 20 | """ |
| 21 | with connection.cursor() as cursor: |
| 22 | cursor.execute(sql, params) |
| 23 | columns = list(map(lambda d: d.name, cursor.description)) |
| 24 | res = cursor.fetchall() |
| 25 | result = list(map(lambda row: dict(list(zip(columns, row))), res)) |
| 26 | cursor.close() |
| 27 | return result |
| 28 | |
| 29 | |
| 30 | def update_execute(sql: str, params): |
no test coverage detected