Python运行时修改业务SQL代码_python_

SELECT * FROM t_demo WHERE tenant_id='xxx' AND is_del=0

SELECT * FROM t_demo WHERE is_del=0

SELECT * FROM t_demo_a WHERE is_del=0

SELECT * FROM t_demo_b WHERE is_del=0

def sql_handle(tenant_id: str): table_name: str = f"t_demo_{tenant_id}" sql: str = f"SELECT * FROM {table_name} WHERE is_del=0"

sql: str = f"SELECT * FROM {table_name} WHERE is_del=0"

def mogrify(self, query, args=None): """ Returns the exact string that is sent to the database by calling the execute() method. This method follows the extension to the DB API 2.0 followed by Psycopg. """ conn = self._get_db() if args is not None: query = query % self._escape_args(args, conn) return query

import pymysql class Cursor(pymysql.cursors.Cursor): def mogrify(self, query: str, args: Union[None, list, dict, tuple] = None) -> str: # 在此可以编写处理还合成的SQL逻辑 mogrify_sql: str = super().mogrify(query, args) # 在此可以编写处理合成后的SQL逻辑 return mogrify_sql class DictCursor(pymysql.cursors.DictCursorMixin, Cursor): """A cursor which returns results as a dictionary""" # 直接修改Cursor类的`mogrify`方法并不会影响到`DictCursor`类，所以我们也要创建一个新的`Cursor`类。

import pymysql.cursors # Connect to the database connection = pymysql.connect( host='localhost', user='user', password='passwd', database='db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor )

with conn.cursor() as cursor: cursor.execute("SELECT * FROM t_demo WHERE is_del=%s", (0, ))

with conn.cursor() as cursor: cursor.execute("SELECT * FROM t_demo WHERE is_del=%(is_del)s", {"is_del": 0})

from contextvars import ContextVar, Token from typing import Any, Dict, Optional, Set context: ContextVar[Dict[str, Any]] = ContextVar("context", default={}) class Context(object): """基础的context调用，支持Type Hints检查""" tenant_id: str replace_table_set: Set[str] def __getattr__(self, key: str) -> Any: value: Any = context.get().get(key) return value def __setattr__(self, key: str, value: Any) -> None: context.get()[key] = value class WithContext(Context): """简单的处理reset token逻辑，和context管理，只用在业务代码""" def __init__(self) -> None: self._token: Optional[Token] = None def __enter__(self) -> "WithContext": self._token = context.set({}) return self def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: if self._token: context.reset(self._token) self._token = None

with WithContext as context: context.tenant_id = "xxx" context.replace_table_set = {"t_demo"} with conn.cursor() as cursor: cursor.execute("SELECT * FROM t_demo WHERE is_del=%s", (0, ))

import pymysql class Cursor(pymysql.cursors.Cursor): def mogrify(self, query: str, args: Union[None, list, dict, tuple] = None) -> str: tenant_id: str = context.tenant_id replace_table_set: Set[str] = context.replace_table_set # 在此可以编写处理还合成的SQL逻辑 mogrify_sql: str = super().mogrify(query, args) # 在此可以编写处理合成后的SQL逻辑 return mogrify_sql

import pymysql class Cursor(pymysql.cursors.Cursor): def mogrify(self, query: str, args: Union[None, list, dict, tuple] = None) -> str: tenant_id: str = context.tenant_id replace_table_set: Set[str] = context.replace_table_set # 简单示例，实际上正则的效率会更好 for replace_table in replace_table_set: if replace_table in query: # 替换表名 query = query.replace(f" {replace_table} ", f" {replace_table}_{tenant_id} ") # 替换查询条件中带有表名的 query = query.replace(f" {replace_table}.", f" {replace_table}_{tenant_id}.") mogrify_sql: str = super().mogrify(query, args) # 在此可以编写处理合成后的SQL逻辑 return mogrify_sql

SELECT * FROM t_demo SELECT * FROM t_demo as demo SELECT * FROM t_other as other LEFT JOIN t_demo demo on demo.xxx==other.xxx

SELECT * FROM `case`

from typing import Dict, Set, Tuple, Union import pymysql import sql_metadata class Cursor(pymysql.cursors.Cursor): def mogrify(self, query: str, args: Union[None, list, dict, tuple] = None) -> str: tenant_id: str = context.tenant_id # 生成一个解析完成的SQL对象 sql_parse: sql_metadata.Parser = sql_metadata.Parser(query) # 新加的一个属性，这里存下需要校验查询条件的表名 check_flag = False where_table_set: Set[str] = context.where_table_set # 该方法会获取到SQL对应的table，返回的是一个table的数组 for table_name in sql_parse.tables: if table_name in where_table_set: if sql_parse.columns_dict: # 该方法会返回SQL对应的字段，其中分为select, join, where等，这里只用到了where for where_column in sql_parse.columns_dict.get("where", []): # 如果连表，里面存的是类似于t_demo.tenant_id，所以要兼容这一个情况 if "tenant_id" in where_column.lower().split("."): check_flag = True break if not check_flag: # 检查不通过就抛错 raise RuntimeError() # 更换表名的逻辑 replace_table_set: Set[str] = context.replace_table_set new_query: str = query for table_name in sql_parse.tables: if table_name in replace_table_set: new_query = "" # tokens存放着解析完的数据，比如SELECT * FROM t_demo解析后是 # [SELECT, *, FROM, t_demo]四个token for token in sql_parse.tokens: # 判断token是否是表名 if token.is_potential_table_name: # 提取规范的表名 parse_table_name: str = token.stringified_token.strip() if parse_table_name in replace_table_set: new_table_name: str = f" {parse_table_name}_{tenant_id}" # next_token代表SQL的下一个字段 if token.next_token.normalized != "AS": # 如果当前表没有设置别名 # 通过AS把替换前的表名设置为新表名的别名，这样一来后面的表名即使没进行更改，也是能读到对应商户ID的表 new_table_name += f" AS {parse_table_name}" query += new_table_name continue # 通过stringi
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：python查看自己安装的所有库并导出的命令_python_
                                                                
下一篇：Python办公自动化处理的10大场景应用示例_python_