1.缘起最近项目在准备搞sass化，sass化有一个特点就是多租户，且每个租户之间的数据都要隔离，对于数据库的隔离方案常见的有数据库隔离，表隔离，字段隔离，目前我只用到表隔离和字段隔离（数据库隔离的原理也是差不多）。 对于字段隔离比较简单，就是查询条件不同而已，比如像下面的sql查询：
select * from t_demo where tenant_id='xxx' and is_del=0
但是为了严谨，需求上需要在执行sql之前检查对应的表是否带上tenant_id的查询字段。
对于表隔离就麻烦了一些，他需要做到在运行的时候根据对应的租户id来处理某个数据表，举个例子，假如有下面这样的一条sql查询:
select * from t_demo where is_del=0
在遇到租户a时，sql查询将变为：
select * from t_demo_a where is_del=0
在遇到租户b时，sql查询将变为：
select * from t_demo_b where is_del=0
如果商户数量固定时，一般在代码里编写if-else来判断就可以了，但是常见的sass化应用的商户是会一直新增的，那么对于这个sql逻辑就会变成这样:
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"
但是这有几个问题，对于orm来说，一开始只创建一个t_demo对应的表对象就可以了，现在却要根据多个商户创建多个表对象，这是不现实的，其次如果是裸写sql，一般会使用ide的检查，而对于这样的sql：
sql: str = f"select * from {table_name} where is_del=0"
ide是没办法进行检查的，当然还有一个最为严重的问题，就是当前的项目已经非常庞大了，如果每个相关表的调用都进行适配更改的话，那工程量就非常庞大了，所以最好的方案就是在引擎库得到用户传过来的sql语句后且还没发送到mysql服务器之前自动的根据商户id更改sql, 而要达到这样的效果，就必须侵入到我们使用的mysql的引擎库，修改里面的方法来兼容我们的需求。
不管是使用dbutils还是sqlalchemy，都可以指定一个引擎库，目前常用的引擎库是pymysql，所以下文都将以pymysql为例进行阐述。
2.侵入库由于必须侵入到我们使用的引擎库，所以我们应该先判断我们需要修改引擎库的哪个方法，在经过源码阅读后，我判定只要更改pymysql.cursors.cursor的mogrify方法:
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
这个方法的作用就是把用户传过来的sql和参数进行整合，生成一个最终的sql,刚好符合我们的需求，于是可以通过继承的思路来创建一个新的属于我们自己的cursor类：
import pymysqlclass 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_sqlclass dictcursor(pymysql.cursors.dictcursormixin, cursor): """a cursor which returns results as a dictionary""" # 直接修改cursor类的`mogrify`方法并不会影响到`dictcursor`类，所以我们也要创建一个新的`cursor`类。
创建好了cursor类后，就需要考虑如何在pymysql中应用我们自定义的cursor类了，一般的mysql连接库都支持我们传入自定义的cursor类，比如pymysql:
import pymysql.cursors# connect to the databaseconnection = pymysql.connect( host='localhost', user='user', password='passwd', database='db', charset='utf8mb4', cursorclass=pymysql.cursors.dictcursor)
我们可以通过cursorclass来指定我们的cursor类,如果使用的库不支持或者是其它原因则需要使用猴子补丁的方法，具体的使用方法见python探针完成调用库的数据提取。
3.获取商户id现在我们已经搞定了在何处修改sql的问题了，接下来就要思考如何在mogrify方法获取到商户id以及那些表要进行替换，一般我们在进行一段代码调用时，有两种传参数的方法， 一种是传数组类型的参数：
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})
目前大多数的项目都存在这两种类型的编写习惯，而引擎库在执行execute时会经过处理后才把参数sql和args传给了mogrify，如果我们是使用字典类型的参数，那么可以在里面嵌入我们需要的参数，并在mogrify里面提取出来，但是使用了数组类型的参数或者是orm库的话就比较难传递参数给mogrify方法了，这时可以通过context隐式的把参数传给mogrify方法，具体的分析和原理可见:python如何使用contextvars模块源码分析。
context的使用方法很简单， 首先是创建一个context封装的类:
from contextvars import contextvar, tokenfrom typing import any, dict, optional, setcontext: 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] = valueclass 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
接下来在业务代码中，通过context传入当前业务对应的参数:
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, ))
然后在mogrify中通过调用context即可获得对应的参数了：
import pymysqlclass 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
4.修改sql现在，万事俱备，只剩下修改sql的逻辑，之前在做别的项目的时候，建的表都是十分的规范，它们是以t_xxx的格式给表命名，这样一来替换表名十分方便，只要进行两次替换就可以兼容大多数情况了，代码如下:
import pymysqlclass 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
但是现在项目的sql规范并不是很好，有些表名还是mysql的关键字，所以靠简单的替换是行不通的，同时这个需求中，一些表只需要字段隔离，需要确保有带上对应的字段查询，这就意味着必须有一个库可以来解析sql，并返回一些数据使我们可以比较方便的知道sql中哪些是表名，哪些是查询字段了。
目前在python中有一个比较知名的sql解析库--sqlparse，它可以通过解析引擎把sql解析成一个python对象,之后我们就可以通过一些语法来判断哪些是sql关键字， 哪些是表名，哪些是查询条件等等。但是这个库只实现一些底层的api，我们需要对他和sql比较了解之后才能实现一些比较完备的功能，比如下面3种常见的sql:
select * from t_demoselect * from t_demo as demoselect * from t_other as other left join t_demo demo on demo.xxx==other.xxx
如果我们要通过sqlparse来提取表名的话就需要处理这3种情况，而我们如果要每一个情况都编写出来的话，那将会非常费心费力，同时也可能存在遗漏的情况，这时就需要用到另外一个库--sql_metadata，这个库是基于sqlparse和正则的解析库，同时提供了大量的常见使用方法的封装，我们通过直接调用对应的函数就能知道sql中有哪些表名，查询字段是什么了。
目前已知这个库有一个缺陷，就是会自动去掉字段的符号， 比如表名为关键字时，我们需要使用`符号把它包起来：
select * from `case`
但在经过sql_metadata解析后得到的表名是case而不是`case`，需要人为的处理，但是我并不觉得这是一个bug，自己不按规范创建表，能怪谁呢。
接下来就可以通过sql_metadata的方法来实现我需要的功能了，在根据需求修改后，代码长这样(说明见注释)：
from typing import dict, set, tuple, unionimport pymysqlimport sql_metadataclass 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 # 通过stringified_token获取的数据会自动带空格，比如`from`得到的会是` from`，这样拼接的时候就不用考虑是否加空格了 new_query += token.stringified_token mogrify_sql: str = super().mogrify(new_query, args) # 在此可以编写处理合成后的sql逻辑 return mogrify_sql
这份代码十分简单，它只做简单介绍，事实上这段逻辑会应用到所有的sql查询中，我们应该要保证这段代码是没问题的，同时不要有太多的性能浪费，所以在使用的时候要考虑到代码拆分和优化。 比如在使用的过程中可以发现，我们的sql转换和检查都是在父类的cursor.mogrify之前进行的，这就意味着不管我们代码逻辑里cursor.execute传的参数是什么，对于同一个代码逻辑来说，传过来的query值是保持不变的，比如下面的代码:
def get_user_info(uid: str) -> dict[str, any]: with conn.cursor() as cursor: cursor.execute("select * from t_user where uid=%(uid)s", {"uid": uid}) return cursor.fetchone() or {}
这段代码中传到cursor.mogrify的query永远为select * from t_user where uid=%(uid)s，有变化的只是args中uid的不同。 有了这样的一个前提条件，那么我们就可以把query的校验结果和转换结果缓存下来，减少每次都需要解析sql再校验造成的性能浪费。至于如何实现缓存则需要根据自己的项目来决定，比如项目中只有几百个sql执行，那么直接用python的dict来存放就可以了，如果项目中执行的sql很多，同时有些执行的频率非常的高，有些执行的频率非常的低，那么可以考虑使用lru来缓存。
以上就是如何在python中根据运行时修改业务sql代码？的详细内容。