Wie ändere ich Business-SQL-Code basierend auf der Laufzeit in Python?

1. Origin

Vor kurzem bereitet sich das Projekt auf die Implementierung von SASS vor. Eines der Merkmale von SASS ist die Mandantenfähigkeit, und die Daten zwischen jedem Mandanten müssen isoliert werden, einschließlich Datenbankisolation, Tabellenisolation und Feldisolation Isolation, derzeit verwende ich nur Tabellenisolation und Feldisolation (das Prinzip der Datenbankisolation ist ähnlich). Die Feldisolierung ist relativ einfach, nur dass die Abfragebedingungen unterschiedlich sind, Zum Beispiel die folgende SQL-Abfrage:

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

Aus Gründen der Genauigkeit muss jedoch überprüft werden, ob die entsprechende Tabelle tenant_id hat code> vor der Ausführung des SQL-Abfragefelds. <code>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 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`类。

创建好了Cursor类后，就需要考虑如何在pymysql中应用我们自定义的Cursor类了，一般的Mysql连接库都支持我们传入自定义的Cursor类，比如pymysql:

import pymysql.cursors
# Connect to the database
connection = 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, 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

接下来在业务代码中，通过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 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

4.修改SQL

现在，万事俱备，只剩下修改SQL的逻辑，之前在做别的项目的时候，建的表都是十分的规范，它们是以t_xxx

Für die Tabellenisolation ist es etwas schwieriger, eine bestimmte Datentabelle während des Betriebs entsprechend der entsprechenden Mandanten-ID zu verarbeiten. Wenn beispielsweise eine SQL-Abfrage wie die folgende vorliegt: 🎜

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

🎜🎜Beim Auftreffen Bei Mieter A lautet die SQL-Abfrage: 🎜🎜

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

🎜🎜 Bei der Begegnung mit Mieter B lautet die SQL-Abfrage: 🎜🎜

SELECT * FROM `case`

🎜Wenn die Anzahl der Händler festgelegt ist, schreiben Sie normalerweise if-else kann zur Beurteilung verwendet werden, aber es werden ständig gängige SASS-Anwendungshändler hinzugefügt, sodass die SQL-Logik wie folgt aussehen wird:🎜<pre class="brush:py;">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) -&gt; 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(&quot;where&quot;, []): # 如果连表，里面存的是类似于t_demo.tenant_id，所以要兼容这一个情况 if &quot;tenant_id&quot; in where_column.lower().split(&quot;.&quot;): 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 = &quot;&quot; # 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&quot; {parse_table_name}_{tenant_id}&quot; # next_token代表SQL的下一个字段 if token.next_token.normalized != &quot;AS&quot;: # 如果当前表没有设置别名 # 通过AS把替换前的表名设置为新表名的别名，这样一来后面的表名即使没进行更改，也是能读到对应商户ID的表 new_table_name += f&quot; AS {parse_table_name}&quot; 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</pre>🎜Aber es gibt mehrere Probleme. Für ORM war es am Anfang genug, um es zu erstellen Nur ein Tabellenobjekt, das <code>t_demo entspricht, ist für mehrere Händler erforderlich. Zweitens ist es unrealistisch, wenn Sie reines SQL schreiben Ein solches SQL: 🎜

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 {}

🎜 IDE kann natürlich auch nicht prüfen, dass das aktuelle Projekt bereits sehr groß ist. Wenn es geändert wird, ist der Arbeitsaufwand groß wird sehr groß sein, daher besteht die beste Lösung darin, die SQL automatisch entsprechend der Händler-ID zu ändern, nachdem die Engine-Bibliothek die vom Benutzer übergebene SQL-Anweisung erhält und bevor sie an den MySQL-Server gesendet wird. Um diesen Effekt zu erzielen, müssen wir in die von uns verwendete MySQL-Engine-Bibliothek eindringen und die darin enthaltenen Methoden so ändern, dass sie mit unseren Anforderungen kompatibel sind. 🎜

🎜Unabhängig davon, ob Sie dbutils oder sqlalchemy verwenden, können Sie eine Engine-Bibliothek angeben. Die derzeit am häufigsten verwendete Engine-Bibliothek ist pymysql. Daher wird im Folgenden alles am Beispiel von pymysql erläutert. 🎜

🎜2. In die Bibliothek eindringen🎜🎜Da wir in die von uns verwendete Engine-Bibliothek eindringen müssen, sollten wir zunächst feststellen, welche Methode der Engine-Bibliothek wir ändern müssen um die mogrify-Methode von pymysql .cursors.Cursor zu ändern: 🎜rrreee🎜Die Funktion dieser Methode besteht darin, das SQL und die vom Benutzer übergebenen Parameter zu integrieren, um ein endgültiges SQL zu generieren , was genau unseren Bedürfnissen entspricht. Wir können also durch Vererbung eine eigene neue Cursor-Klasse erstellen: 🎜rrreee🎜Nachdem wir die Cursor-Klasse erstellt haben, müssen wir überlegen, wie das geht Verwenden Sie es in . Unsere benutzerdefinierte <code>Cursor-Klasse wird in pymysql verwendet. Die allgemeine Mysql-Verbindungsbibliothek unterstützt uns bei der Übergabe des benutzerdefinierten Cursor Klasse, wie zum Beispiel <code>pymysql:🎜rrreee🎜Wir können unsere Cursor-Klasse über cursorclass angeben, wenn die verwendete Bibliothek sie nicht unterstützt oder aus anderen Gründen Sie müssen die Monkey-Patch-Methode verwenden. Informationen zu spezifischen Verwendungsmethoden finden Sie unter Python-Probe, um die Datenextraktion aus der aufrufenden Bibliothek abzuschließen. 🎜🎜3. Holen Sie sich die Händler-ID🎜🎜Da wir nun herausgefunden haben, wo wir die SQL ändern können, müssen wir darüber nachdenken, wie wir die Händler-ID in der mogrify-Methode erhalten und welche Tabellen erforderlich sind Wenn wir einen Code aufrufen, gibt es im Allgemeinen zwei Möglichkeiten, Parameter zu übergeben: 🎜rrreee🎜 Die andere besteht darin, Parameter vom Typ Wörterbuch zu übergeben: 🎜rrreee🎜 Die meisten Projekte verfügen derzeit über diese beiden Methoden . Diese Art von Schreibgewohnheit und die Engine-Bibliothek übergeben die Parameter sql und args nach der Verarbeitung an , wenn <code>execute mogrify: Wenn wir Parameter vom Wörterbuchtyp verwenden, können wir die benötigten Parameter darin einbetten und in mogrify extrahieren. Verwenden Sie jedoch Parameter vom Array-Typ oder die ORM-Bibliothek. Es wird schwieriger sein, Parameter zu übergeben Die mogrify-Methode kann zu diesem Zeitpunkt implizit über context an die Methode mogrify übergeben werden : So verwenden Sie die Quellcodeanalyse des contextvars-Moduls in Python. 🎜🎜Die Methode zur Verwendung von context ist sehr einfach. Erstellen Sie zunächst eine durch context gekapselte Klasse: 🎜rrreee🎜Als nächstes übergeben Sie im Geschäftscode die entsprechenden Daten von das aktuelle Geschäft durch Kontext , alles ist fertig, der Rest besteht darin, die Logik von SQL zu ändern. Als ich zuvor an anderen Projekten gearbeitet habe, waren die Tabellen im Format t_xxx sehr standardisiert. Dies macht es sehr praktisch, die Tabellennamen zu ersetzen. Solange zwei Ersetzungen vorgenommen werden, ist dies mit den meisten Situationen kompatibel. Der Code lautet wie folgt: 🎜

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

但是现在项目的SQL规范并不是很好，有些表名还是MySQL的关键字，所以靠简单的替换是行不通的，同时这个需求中，一些表只需要字段隔离，需要确保有带上对应的字段查询，这就意味着必须有一个库可以来解析SQL，并返回一些数据使我们可以比较方便的知道SQL中哪些是表名，哪些是查询字段了。

目前在Python中有一个比较知名的SQL解析库--sqlparse，它可以通过解析引擎把SQL解析成一个Python对象,之后我们就可以通过一些语法来判断哪些是SQL关键字， 哪些是表名，哪些是查询条件等等。但是这个库只实现一些底层的API，我们需要对他和SQL比较了解之后才能实现一些比较完备的功能，比如下面3种常见的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

如果我们要通过sqlparse来提取表名的话就需要处理这3种情况，而我们如果要每一个情况都编写出来的话，那将会非常费心费力，同时也可能存在遗漏的情况，这时就需要用到另外一个库--sql_metadata，这个库是基于sqlparse和正则的解析库，同时提供了大量的常见使用方法的封装，我们通过直接调用对应的函数就能知道SQL中有哪些表名，查询字段是什么了。

目前已知这个库有一个缺陷，就是会自动去掉字段的符号， 比如表名为关键字时，我们需要使用`符号把它包起来：

SELECT * FROM `case`

但在经过sql_metadata解析后得到的表名是case而不是`case`，需要人为的处理，但是我并不觉得这是一个BUG，自己不按规范创建表，能怪谁呢。

接下来就可以通过sql_metadata的方法来实现我需要的功能了，在根据需求修改后，代码长这样(说明见注释)：

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
                    # 通过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来缓存。

Das obige ist der detaillierte Inhalt vonWie ändere ich Business-SQL-Code basierend auf der Laufzeit in Python?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!