扫码关注官方微信

扫码下载APP
返回顶部
首页 > 资讯 > 后端开发 > Python >Python运行时修改业务SQL代码
  • 309

0
分享到

Python运行时修改业务SQL代码

2024-04-02 19:04:59 309人浏览 薄情痞子

Python 官方文档：入门教程 => 点击学习

摘要

目录前记1.缘起2.侵入库3.获取商户ID4.修改sql前记 在项目的演变过程中，有时可能会诞生一些需要奇怪的临时需求，这些需求会涉及到所有的SQL，但开发时间上却不允许整个项目的所

前记

在项目的演变过程中，有时可能会诞生一些需要奇怪的临时需求，这些需求会涉及到所有的SQL，但开发时间上却不允许整个项目的所有SQL进行重写，比如控制不同的人访问表的权限，或者是我面对的SASS化需求，这时就需要在运行时根据对应的条件来修改SQL语句。

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 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的格式给表命名，这样一来替换表名十分方便，只要进行两次替换就可以兼容大多数情况了，代码如下:

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

到此这篇关于Python运行时修改业务SQL代码的文章就介绍到这了,更多相关Python 修改代码内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: Python运行时修改业务SQL代码

本文链接: https://lsjlt.com/news/118903.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

猜你喜欢
  • Python运行时修改业务SQL代码
    目录前记1.缘起2.侵入库3.获取商户ID4.修改SQL前记 在项目的演变过程中，有时可能会诞生一些需要奇怪的临时需求，这些需求会涉及到所有的SQL，但开发时间上却不允许整个项目的所...
    99+
    2024-04-02
  • Python运行时修改业务SQL代码怎么写
    这篇“Python运行时修改业务SQL代码怎么写”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“Python运行时修改业务SQ...
    99+
    2023-07-02
  • python获取代码运行时间
    有的时候，操作大文件，或者取数，要很久，我们给脚本首尾添加一段代码就知道，这段代码整体的大致运行时间了。 import time start =time.clock() #中间写上代码块 end = time.clock() ...
    99+
    2023-01-31
    代码 时间 python
  • git修改临时代码
    在开发过程中，经常会遇到需要调试或者测试某个特定的功能，这就需要我们去修改代码来满足我们的需求。但是这时候我们又不想影响到其他人的工作，或者导致一些意外后果。这时候我们需要使用git来修改临时代码，以达到不破坏原代码并能够单独处理的效果。G...
    99+
    2023-10-22
  • 云服务器运行python代码
    以下是一个简单的 Python 云服务器示例代码,它将运行在 AWS 上,并创建一个名为 "cloud" 的文件夹,将您的数据保存在这个文件夹中。 ```python import csharp 设置 Python 运行环境 with o...
    99+
    2023-10-26
    代码 服务器 python
  • python怎么运行代码
    python运行代码的方法：使用快捷键【win+R 键】打开运行窗口，然后输入：“cmd”，点击确定打开dos窗口。切换到你需要运行的文件所在的盘。再打开你需要运行的文件夹中。最后输入：python ***.py（程序文件的名字）...
    99+
    2024-04-02
  • python中的代码运行时间获取方式
    目录第一种第二种第三种第四种我们知道为了提高代码的运行速度，我们需要对书写的python代码进行性能测试，而代码性能的高低的直接反馈是电脑运行代码所需要的时间。 这里将介绍四种常用的...
    99+
    2024-04-02
  • 如何在 Windows 上实时运行 Python Leetcode 代码？
    Python是一种流行的编程语言，其易于学习和使用，成为许多程序员的首选语言。在Leetcode上练习算法问题是提高编程技能的一个好方法。在Windows上运行Python Leetcode代码需要一些特定的设置。本文将介绍如何在Windo...
    99+
    2023-09-24
    leetcode 实时 windows
  • chatgpt赋能python：如何同时运行两个Python代码
    如何同时运行两个Python代码 Python是一种广泛使用的高级编程语言，广泛应用于数据科学、人工智能、网络开发等领域。在学习和使用Python时，我们经常需要同时运行多个代码文件。本文将介绍如何使...
    99+
    2023-09-14
    python chatgpt 开发语言 计算机
  • spyder怎么运行python代码
    要在Spyder中运行Python代码，您可以按照以下步骤操作：1. 打开Spyder：在Windows系统上，可以从开始菜单中找到...
    99+
    2023-10-12
    spyder python
  • jupyter如何运行python代码
    要在Jupyter中运行Python代码，您可以按照以下步骤进行操作：1. 打开Jupyter Notebook：在命令行中输入ju...
    99+
    2023-10-19
    jupyter python
  • pycharm怎么运行python代码
    使用 pycharm 运行 python 代码的步骤：创建 python 项目。创建脚本或程序。单击“运行”按钮或按快捷键运行脚本。使用调试器在调试模式下运行代码。其他运行选项包括：运行...
    99+
    2024-04-03
    linux python pycharm
  • Python获取代码运行时间的几种方法
    Python获取代码运行时间的几种方法1、方法一：#python 的标准库手册推荐在任何情况下尽量使用time.clock().#只计算了程序运行CPU的时间，返回值是浮点数import timestart =time.clock()#中...
    99+
    2023-01-31
    几种方法 代码 时间
  • 云服务器一直运行python代码
    打开控制面板：在控制面板中，打开“管理工具”选项。 选择“服务”：在服务列表中，找到“Amazon S3”服务。 停止该服务：单击“停止”按钮，然后选择“已禁用”选项。 重启服务器：选择“服务”选项卡，然后单击“启动”按钮。 如果您无...
    99+
    2023-10-27
    代码 服务器 python
  • 如何修改php的最长运行时间
    这篇文章主要介绍了如何修改php的最长运行时间，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。修改php的最长运行时间的方法是，编辑php.ini配置文件，修改max_exec...
    99+
    2023-06-15
  • Python在线运行代码助手
    Python代码运行助手可以让你在线输入Python代码，然后通过本机运行的一个Python脚本来执行代码。原理如下： 在网页输入代码： 点击Run按钮，代码被发送到本机正在运行的Python代码运行助手...
    99+
    2022-06-04
    在线 助手 代码
  • 如何用iPad运行Python代码
    这篇文章给大家介绍如何用iPad运行Python代码，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。其实，不只是iPad，手机也可以。痛点我组织过几次线下编程工作坊，带着同学们用Python处理数据科学问题。其中最让人头...
    99+
    2023-06-17
  • python爱心代码怎么运行
    要运行 python 爱心代码，步骤如下：安装 python 解释器。创建一个 python 文件并复制代码。保存文件并命名为 .py 扩展名。打开终端或命令提示符。导航到 python...
    99+
    2024-04-02
  • 如何在Python退出时强制运行一段代码
    这篇文章将为大家详细讲解有关如何在Python退出时强制运行一段代码，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。设想这样一个场景，你要给一个项目开发测试程序，程序开始运行的时候，会创建初始...
    99+
    2023-06-03
  • python怎么换行而不运行代码
    在Python中，可以使用特殊字符`\n`来表示换行。当你想要在代码中换行而不让其运行，可以在需要换行的地方添加`\n`。例如：``...
    99+
    2023-08-17
    python
软考高级职称资格查询
热门wiki
近期文章
推荐阅读
热门问答
热门标签
编程网，编程工程师的家园，是目前国内优秀的开源技术社区之一，形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容，为IT开发者提供了一个发现、使用、并交流开源技术的平台。

  • 官方手机版

  • 微信公众号

  • 商务合作