欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

利用Python實(shí)現(xiàn)一個(gè)類似MybatisPlus的簡(jiǎn)易SQL注解

 更新時(shí)間:2025年08月05日 10:54:22   作者:用戶57690530801  
在實(shí)際開發(fā)中,根據(jù)業(yè)務(wù)拼接SQL所需要考慮的內(nèi)容太多了,于是,有沒有一種辦法,可以像MyBatisPlus一樣通過配置注解實(shí)現(xiàn)SQL注入呢?本文給大家介紹了如何利用Python實(shí)現(xiàn)一個(gè)類似MybatisPlus的簡(jiǎn)易SQL注解,需要的朋友可以參考下

前言

在實(shí)際開發(fā)中,根據(jù)業(yè)務(wù)拼接SQL所需要考慮的內(nèi)容太多了。于是,有沒有一種辦法,可以像MyBatisPlus一樣通過配置注解實(shí)現(xiàn)SQL注入呢?

就像是:

@mybatis.select("select * from user where id = #{id}")
def get_user(id): ...

那可就降低了好多工作量。

P.S.:本文并不希望完全復(fù)現(xiàn)MyBatisPlus的所有功能,能夠基本配置SQL注解就基本能夠完成大部分工作了。

實(shí)現(xiàn)思路

那我們這么考慮:

  1. 首先,我們需要定義一個(gè)類,類中給一個(gè)或者多個(gè)裝飾器;
  2. 我們先在類內(nèi)定義一個(gè)字符串,這個(gè)字符串能夠配置到指定的DTO類,用于存儲(chǔ)結(jié)果;
  3. 我們針對(duì)裝飾器中的SQL字符串進(jìn)行解析,解析到其中的變量個(gè)數(shù)與名稱;
  4. 我們針對(duì)被裝飾的函數(shù)進(jìn)行解析,與SQL變量進(jìn)行匹配;
  5. 替換變量;
  6. 執(zhí)行SQL;

聽起來并不難。我們一步步來。

定義一個(gè)類

首先定義:

# dto/student.py
class Student:
    def __init__(self, name, age):
        self.name = name
        self.age = age

為了簡(jiǎn)化操作,這個(gè)類就不放在任意位置了,直接放在dto文件夾下,后續(xù)導(dǎo)入這個(gè)類也就直接從dto文件夾中引入,就不考慮做這個(gè)包名定位的接口了。

當(dāng)然,為了更方便后續(xù)的操作,我們需要在dto文件夾中定義一個(gè)__init__.py文件,用于對(duì)外暴露這個(gè)類:

# dto/__init__.py
from dto.student import Student
__all__ = ["Student"]

最后呢,我們?yōu)榱朔奖氵@個(gè)類的序列化,讓他能夠變成dict類型,加一些魔法函數(shù):

# dto/student.py
class Student:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    def __iter__(self):
        for key, value in self.__dict__.items():
            yield key, value
    def __getitem__(self, key):
        return getattr(self, key)
    def keys(self):
        return self.__dict__.keys()

當(dāng)然,一個(gè)項(xiàng)目里面肯定不止這一個(gè)返回結(jié)果,所以各位也可以這么操作:

# dto/common.py
class CommonResult:
    def __init__(self): ...
    def __iter__(self):
        for key, value in self.__dict__.items():
            yield key, value
    def __getitem__(self, key):
        return getattr(self, key)
    def keys(self):
        return self.__dict__.keys()
# dto/student.py
from dto.common import CommonResult
class Student(CommonResult):
    def __init__(self, name, age):
        self.name = name
        self.age = age

至于實(shí)際業(yè)務(wù)中還有很多復(fù)雜的聯(lián)立等操作需要新的類,受限于篇幅,就不展開了。如果能夠把本篇看懂的話,相信各位也沒什么其他的困難了。

然后開始手?jǐn)]這個(gè)微型框架

# db/common.py
from pydantic import BaseModel, Field

class DBManager(BaseModel):
  base_type: str = Field(..., description="數(shù)據(jù)庫表名")
  link: str = Field(..., description="數(shù)據(jù)庫連接地址")
  local_generator: Any = Field(..., description="實(shí)體類實(shí)例化解析生成器")
  def search(query_template): ...

在這里呢,我們定義了一個(gè)DBManager作為父類,要求后面的子類必須有:

  • str類型的base_type,表示返回結(jié)果類的名稱;
  • str類型的link,表示數(shù)據(jù)庫連接地址;
  • Any類型的local_generator,表示實(shí)體類實(shí)例化解析生成器,- 任意返回值的query方法,用于執(zhí)行SQL。

為什么一定要用BaseModel定義?直接定義self.xxx不好嗎?

因?yàn)檫@樣會(huì)看起來代碼量很大(逃)

看著差不多。

根據(jù)字符串獲取到所定義的DTO類

考慮到實(shí)際上我們所有的方法都需要特定到具體的位置,所以這個(gè)方法還是直接寫到DBManager類中,這樣子類就不需要再重寫了。

# db/common.py
from pydantic import BaseModel, Field

class DBManager(BaseModel):
    base_type: str = Field(..., description="數(shù)據(jù)庫表名")
    link: str = Field(..., description="數(shù)據(jù)庫連接地址")
    local_generator: Any = Field(..., description="實(shí)體類實(shí)例化解析生成器")

    def search(query_template): ...

    def import_class_from_package(self, package_name, class_name):
        # 根據(jù)包名獲得`DTO`包
        _package = importlib.import_module(package_name)
        # 檢測(cè)是不是有這么個(gè)類
        if class_name not in _package.__all__:
            raise ImportError(f"{class_name} not found in {package_name}")
        # 有就拿著
        cls = getattr(_package, class_name)
        # 返回這個(gè)類
        if cls is not None:
            return cls
        else:
            raise ImportError(f"{class_name} not found in {package_name}")

這樣子類就可以調(diào)用這個(gè)方法獲得所需的類了。

構(gòu)建返回結(jié)果

既然都已經(jīng)能夠動(dòng)態(tài)導(dǎo)入類了,那我把返回結(jié)果導(dǎo)入到Student中,沒問題吧?

其中需要注意的是,我這邊采用的數(shù)據(jù)庫驅(qū)動(dòng)是sqlalchemy,所以構(gòu)造返回結(jié)果所需要的參數(shù)是sqlalchemyRow類型。

同樣的,為了減少子類重寫的代碼量,直接在父類給出來:

# db/common.py
from pydantic import BaseModel, Field
from sqlalchemy.engine.row import Row

class DBManager(BaseModel):
    base_type: str = Field(..., description="數(shù)據(jù)庫表名")
    link: str = Field(..., description="數(shù)據(jù)庫連接地址")
    local_generator: Any = Field(..., description="實(shí)體類實(shí)例化解析生成器")

    def search(query_template): ...
    # 為了方便看,省略掉細(xì)節(jié)
    def import_class_from_package(self, package_name, class_name): ...

    def build_obj(self, row: Row):
        return self.local_generator(**row._asdict()) if self.local_generator else None

裝飾器

那么接下來就是重頭戲了,怎么定義這個(gè)裝飾器。

我們先構(gòu)建一個(gè)子類:

# db/student.py
class StudentDBManager(DBManager):
    base_type: ClassVar[str] = "Student"
    link: ClassVar[str] = 'sqlite:///school.db'
    local_generator: ClassVar[Any] = None

    """
    自定義PyMyBatis
    """
    def __init__(self):
        StudentDBManager.local_generator = self.import_class_from_package("dto", self.base_type)

在這里,首先需要注意的是,需要用ClassVar修飾,將變量名定義為類內(nèi)成員變量,否則無法使用self.xxx訪問。

其次,我們利用base_type指定返回值對(duì)應(yīng)的DTO類、link指定數(shù)據(jù)庫連接地址,local_generator指定實(shí)體類實(shí)例化解析生成器。

在這個(gè)類實(shí)例化的過程中,我們還需要進(jìn)一步構(gòu)建local_generator,也就是動(dòng)態(tài)執(zhí)行from xxx import xxx

然后定義一個(gè)裝飾器:

def query(query_template: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

這可以算得上是比較基礎(chǔ)的模板了。至于之后怎么改,管他呢,先套公式。

在這里,我們首先定義的裝飾器是decorator,沒有參數(shù);其次再用query裝飾器包裝,從而給無參的裝飾器給一個(gè)參數(shù),從而接收一個(gè)SQL字符串參數(shù)。

好的,我們?cè)龠M(jìn)一步。

解析字符串,獲得變量

首先當(dāng)然是解析SQL字符串,獲得變量。如何做呢?為了簡(jiǎn)便,這里直接采用正則匹配的方式:

def query(self, query_template):
    def decorator(func):
        # 解析 SQL 中的 #{變量} 語法
        param_pattern = re.compile(r"#{(\w+)}")
        required_params = set(param_pattern.findall(query_template))
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

沒啥問題。

接下來,調(diào)用的時(shí)候,我們需要檢測(cè)是否完整給出了SQL字符串所需的參數(shù)。

我們考慮到,如果但凡SQL中的參數(shù)有變化,方法就會(huì)有變化,因此每個(gè)SQL都有一個(gè)方法也太麻煩了。主要是這么多相似的方法起方法名太煩了

所以,直接上反射,獲取 調(diào)用 的時(shí)侯傳入的參數(shù)。

值得注意的是,這里說的是 調(diào)用 的時(shí)候。因?yàn)?code>Python中 定義 方法的時(shí)候可以使用**kargs傳入多個(gè)參數(shù),但是如果反射直接獲取到 定義 的參數(shù),將會(huì)只有一個(gè)kargs,這顯然不是我們所希望的。

所以,再加一些:

def query(self, query_template):
    def decorator(func):
        # 解析 SQL 中的 #{變量} 語法
        param_pattern = re.compile(r"#{(\w+)}")
        required_params = set(param_pattern.findall(query_template))
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 獲取函數(shù)的參數(shù)簽名
            sig = inspect.signature(func)
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()
            # 提取傳遞的參數(shù),包括 **kwargs 中的參數(shù)
            provided_params = set(bound_args.arguments.keys()) | set(kwargs.keys())
            # 檢查缺失的參數(shù)
            missing_params = required_params - provided_params
            if missing_params:
                raise ValueError(f"Missing required parameters: {', '.join(missing_params)}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

這下應(yīng)該就能夠適配到所有的SQL情況了。

SQL字符串拼接

接下來就是直接替換值了。但是,拼接真的就是對(duì)的嗎?我們不光是需要考慮不同的變量有著不同的植入格式,同時(shí)也需要考慮到植入過程中可能的SQL注入問題。

所以,我們就直接采用sqlalchemytext函數(shù),對(duì)SQL進(jìn)行拼接與賦值。

def query(self, query_template):
    def decorator(func):
        # 解析 SQL 中的 #{變量} 語法
        param_pattern = re.compile(r"#{(\w+)}")
        required_params = set(param_pattern.findall(query_template))
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 獲取函數(shù)的參數(shù)簽名
            sig = inspect.signature(func)
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()
            # 提取傳遞的參數(shù),包括 **kwargs 中的參數(shù)
            provided_params = set(bound_args.arguments.keys()) | set(kwargs.keys())
            # 檢查缺失的參數(shù)
            missing_params = required_params - provided_params
            if missing_params:
                raise ValueError(f"Missing required parameters: {', '.join(missing_params)}")
            # 構(gòu)建 SQL 語句,并考慮不同類型的數(shù)據(jù)格式
            sql_query = text(query_template.replace("#{", ":").replace("}", ""))
            print(f"Executing SQL: {sql_query}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

好了,到這一步也就基本完成了。最后,我們根據(jù)數(shù)據(jù)庫存儲(chǔ)數(shù)據(jù)的特點(diǎn),最后修整一下查詢的格式細(xì)節(jié),就可以了:

def query(self, query_template):
    def decorator(func):
        # 解析 SQL 中的 #{變量} 語法
        param_pattern = re.compile(r"#{(\w+)}")
        required_params = set(param_pattern.findall(query_template))
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 獲取函數(shù)的參數(shù)簽名
            sig = inspect.signature(func)
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()
            # 提取傳遞的參數(shù),包括 **kwargs 中的參數(shù)
            provided_params = set(bound_args.arguments.keys()) | set(kwargs.keys())
            # 檢查缺失的參數(shù)
            missing_params = required_params - provided_params
            if missing_params:
                raise ValueError(f"Missing required parameters: {', '.join(missing_params)}")
            # 構(gòu)建 SQL 語句,并考慮不同類型的數(shù)據(jù)格式
            sql_query = text(query_template.replace("#{", ":").replace("}", ""))
            print(f"Executing SQL: {sql_query}")
            params = bound_args.arguments.copy()
            for key, value in params.items():
                if isinstance(value, datetime):
                    params[key] = value.strftime('%Y-%m-%d')
            engine = create_engine(self.link)
            with engine.connect() as conn:
                result = conn.execute(sql_query, params)
                search_result = [self.create_item_obj(row) for row in result]
            return search_result
        return wrapper
    return decorator

就是這樣,我們就完成了這樣一個(gè)裝飾器。

使用裝飾器

使用過程,其實(shí)就可以類比@Service中的調(diào)用了。而如果拿Python舉例的話,其實(shí)更像Flaskapp.route。于是我們可以這么使用:

sbd = StudentDBManager()
@sbd.query("SELECT * FROM student WHERE id = #{id}")
def find_student_by_id(**kargs): ...

這也就實(shí)現(xiàn)了一個(gè)方法。

當(dāng)然,他也沒那么智能。雖然寫起來是這樣,但是依然相當(dāng)于:

sbd = StudentDBManager()
@sbd.query("SELECT * FROM student WHERE id = #{id}")
def find_student_by_id(id: str): ...

只是說,我們并不需要重復(fù)地去寫驅(qū)動(dòng)罷了。

以上就是利用Python實(shí)現(xiàn)一個(gè)類似MybatisPlus的簡(jiǎn)易SQL注解的詳細(xì)內(nèi)容,更多關(guān)于Python簡(jiǎn)易SQL注解的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • Django celery異步任務(wù)實(shí)現(xiàn)代碼示例

    Django celery異步任務(wù)實(shí)現(xiàn)代碼示例

    這篇文章主要介紹了Django celery異步任務(wù)實(shí)現(xiàn)代碼示例,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下
    2020-11-11
  • Python標(biāo)準(zhǔn)庫之time庫的使用教程詳解

    Python標(biāo)準(zhǔn)庫之time庫的使用教程詳解

    這篇文章主要介紹了Python的time庫的使用教程,文中有非常詳細(xì)的代碼示例,對(duì)正在學(xué)習(xí)python基礎(chǔ)的小伙伴們有非常好的幫助,需要的朋友可以參考下
    2022-04-04
  • 在VS2017中用C#調(diào)用python腳本的實(shí)現(xiàn)

    在VS2017中用C#調(diào)用python腳本的實(shí)現(xiàn)

    這篇文章主要介紹了在VS2017中用C#調(diào)用python腳本的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2019-07-07
  • Python和OpenCV自制訪客識(shí)別程序

    Python和OpenCV自制訪客識(shí)別程序

    這篇文章主要為大家詳細(xì)介紹了如何使用Python和OpenCV自制訪客識(shí)別程序,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下
    2024-11-11
  • 如何將你的應(yīng)用遷移到Python3的三個(gè)步驟

    如何將你的應(yīng)用遷移到Python3的三個(gè)步驟

    這篇文章主要介紹了如何將你的應(yīng)用遷移到Python3的三個(gè)步驟,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2019-12-12
  • python腳本爬取字體文件的實(shí)現(xiàn)方法

    python腳本爬取字體文件的實(shí)現(xiàn)方法

    這篇文章主要給大家介紹了利用python腳本爬取字體文件的實(shí)現(xiàn)方法,文中分享了爬取兩個(gè)不同網(wǎng)站的示例代碼,相信對(duì)大家具有一定的參考價(jià)值,需要的朋友們下面來一起看看吧。
    2017-04-04
  • Python3 把一個(gè)列表按指定數(shù)目分成多個(gè)列表的方式

    Python3 把一個(gè)列表按指定數(shù)目分成多個(gè)列表的方式

    今天小編就為大家分享一篇Python3 把一個(gè)列表按指定數(shù)目分成多個(gè)列表的方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧
    2019-12-12
  • xadmin使用formfield_for_dbfield函數(shù)過濾下拉表單實(shí)例

    xadmin使用formfield_for_dbfield函數(shù)過濾下拉表單實(shí)例

    這篇文章主要介紹了xadmin使用formfield_for_dbfield函數(shù)過濾下拉表單實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧
    2020-04-04
  • python實(shí)現(xiàn)簡(jiǎn)單遺傳算法

    python實(shí)現(xiàn)簡(jiǎn)單遺傳算法

    這篇文章主要介紹了python如何實(shí)現(xiàn)簡(jiǎn)單遺傳算法,幫助大家更好的利用python進(jìn)行數(shù)據(jù)分析,感興趣的朋友可以了解下
    2020-09-09
  • Python線程threading模塊用法詳解

    Python線程threading模塊用法詳解

    這篇文章主要介紹了Python線程threading模塊用法,結(jié)合實(shí)例形式總結(jié)分析了Python線程threading模塊基本功能、原理、相關(guān)函數(shù)使用方法與操作注意事項(xiàng),需要的朋友可以參考下
    2020-02-02

最新評(píng)論