利用Python實現(xiàn)一個類似MybatisPlus的簡易SQL注解
前言
在實際開發(fā)中，根據(jù)業(yè)務(wù)拼接SQL所需要考慮的內(nèi)容太多了。于是，有沒有一種辦法，可以像MyBatisPlus一樣通過配置注解實現(xiàn)SQL注入呢？
就像是：
@mybatis.select("select * from user where id = #{id}")
def get_user(id): ...
那可就降低了好多工作量。
P.S.：本文并不希望完全復(fù)現(xiàn)MyBatisPlus的所有功能，能夠基本配置SQL注解就基本能夠完成大部分工作了。
實現(xiàn)思路
那我們這么考慮：
- 首先，我們需要定義一個類，類中給一個或者多個裝飾器；
- 我們先在類內(nèi)定義一個字符串，這個字符串能夠配置到指定的
DTO類，用于存儲結(jié)果； - 我們針對裝飾器中的
SQL字符串進行解析，解析到其中的變量個數(shù)與名稱； - 我們針對被裝飾的函數(shù)進行解析，與
SQL變量進行匹配； - 替換變量；
- 執(zhí)行
SQL；
聽起來并不難。我們一步步來。
定義一個類
首先定義：
# dto/student.py
class Student:
def __init__(self, name, age):
self.name = name
self.age = age
為了簡化操作，這個類就不放在任意位置了，直接放在dto文件夾下，后續(xù)導(dǎo)入這個類也就直接從dto文件夾中引入，就不考慮做這個包名定位的接口了。
當然，為了更方便后續(xù)的操作，我們需要在dto文件夾中定義一個__init__.py文件，用于對外暴露這個類：
# dto/__init__.py from dto.student import Student __all__ = ["Student"]
最后呢，我們?yōu)榱朔奖氵@個類的序列化，讓他能夠變成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()
當然，一個項目里面肯定不止這一個返回結(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
至于實際業(yè)務(wù)中還有很多復(fù)雜的聯(lián)立等操作需要新的類，受限于篇幅，就不展開了。如果能夠把本篇看懂的話，相信各位也沒什么其他的困難了。
然后開始手擼這個微型框架
# 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="實體類實例化解析生成器") def search(query_template): ...
在這里呢，我們定義了一個DBManager作為父類，要求后面的子類必須有：
str類型的base_type，表示返回結(jié)果類的名稱；str類型的link，表示數(shù)據(jù)庫連接地址；Any類型的local_generator，表示實體類實例化解析生成器，- 任意返回值的query方法，用于執(zhí)行SQL。
為什么一定要用BaseModel定義？直接定義self.xxx不好嗎？
因為這樣會看起來代碼量很大（逃）
看著差不多。
根據(jù)字符串獲取到所定義的DTO類
考慮到實際上我們所有的方法都需要特定到具體的位置，所以這個方法還是直接寫到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="實體類實例化解析生成器")
def search(query_template): ...
def import_class_from_package(self, package_name, class_name):
# 根據(jù)包名獲得`DTO`包
_package = importlib.import_module(package_name)
# 檢測是不是有這么個類
if class_name not in _package.__all__:
raise ImportError(f"{class_name} not found in {package_name}")
# 有就拿著
cls = getattr(_package, class_name)
# 返回這個類
if cls is not None:
return cls
else:
raise ImportError(f"{class_name} not found in {package_name}")
這樣子類就可以調(diào)用這個方法獲得所需的類了。
構(gòu)建返回結(jié)果
既然都已經(jīng)能夠動態(tài)導(dǎo)入類了，那我把返回結(jié)果導(dǎo)入到Student中，沒問題吧？
其中需要注意的是，我這邊采用的數(shù)據(jù)庫驅(qū)動是sqlalchemy，所以構(gòu)造返回結(jié)果所需要的參數(shù)是sqlalchemy的Row類型。
同樣的，為了減少子類重寫的代碼量，直接在父類給出來：
# 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="實體類實例化解析生成器")
def search(query_template): ...
# 為了方便看，省略掉細節(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òu)建一個子類：
# 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指定返回值對應(yīng)的DTO類、link指定數(shù)據(jù)庫連接地址，local_generator指定實體類實例化解析生成器。
在這個類實例化的過程中，我們還需要進一步構(gòu)建local_generator，也就是動態(tài)執(zhí)行from xxx import xxx。
然后定義一個裝飾器：
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裝飾器包裝，從而給無參的裝飾器給一個參數(shù)，從而接收一個SQL字符串參數(shù)。
好的，我們再進一步。
解析字符串，獲得變量
首先當然是解析SQL字符串，獲得變量。如何做呢？為了簡便，這里直接采用正則匹配的方式：
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)用的時候，我們需要檢測是否完整給出了SQL字符串所需的參數(shù)。
我們考慮到，如果但凡SQL中的參數(shù)有變化，方法就會有變化，因此每個SQL都有一個方法也太麻煩了。主要是這么多相似的方法起方法名太煩了
所以，直接上反射，獲取 調(diào)用 的時侯傳入的參數(shù)。
值得注意的是，這里說的是 調(diào)用 的時候。因為Python中 定義 方法的時候可以使用**kargs傳入多個參數(shù)，但是如果反射直接獲取到 定義 的參數(shù)，將會只有一個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字符串拼接
接下來就是直接替換值了。但是，拼接真的就是對的嗎？我們不光是需要考慮不同的變量有著不同的植入格式，同時也需要考慮到植入過程中可能的SQL注入問題。
所以，我們就直接采用sqlalchemy的text函數(shù)，對SQL進行拼接與賦值。
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ù)庫存儲數(shù)據(jù)的特點，最后修整一下查詢的格式細節(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
就是這樣，我們就完成了這樣一個裝飾器。
使用裝飾器
使用過程，其實就可以類比@Service中的調(diào)用了。而如果拿Python舉例的話，其實更像Flask的app.route。于是我們可以這么使用：
sbd = StudentDBManager()
@sbd.query("SELECT * FROM student WHERE id = #{id}")
def find_student_by_id(**kargs): ...
這也就實現(xiàn)了一個方法。
當然，他也沒那么智能。雖然寫起來是這樣，但是依然相當于：
sbd = StudentDBManager()
@sbd.query("SELECT * FROM student WHERE id = #{id}")
def find_student_by_id(id: str): ...
只是說，我們并不需要重復(fù)地去寫驅(qū)動罷了。
以上就是利用Python實現(xiàn)一個類似MybatisPlus的簡易SQL注解的詳細內(nèi)容，更多關(guān)于Python簡易SQL注解的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
在面向?qū)ο蟪绦蛟O(shè)計中，除了封裝和繼承特性外，多態(tài)也是一個非常重要的特性。讓具有不同功能的函數(shù)可以使用相同的函數(shù)名，這樣就可以用一個函數(shù)名調(diào)用不同內(nèi)容(功能)的函數(shù)2021-06-06
Pytorch實現(xiàn)常用乘法算子TensorRT的示例代碼
pytorch 用于訓練，TensorRT用于推理是很多AI應(yīng)用開發(fā)的標配。大家往往更加熟悉 pytorch 的算子，而不太熟悉TensorRT的算子。本文介紹了Pytorch中常用乘法的TensorRT實現(xiàn)，感興趣的可以了解一下2022-06-06
這篇文章主要介紹了利用Python如何實時檢測自身內(nèi)存占用的相關(guān)資料，文中通過示例代碼介紹的非常詳細，對大家學習或者使用Python具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2020-05-05
這篇文章主要介紹了PyQt5中向單元格添加控件的方法示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-03-03
這篇文章主要介紹了完美解決ARIMA模型中plot_acf畫不出圖的問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06
本文主要介紹了pandas對齊運算的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-10-10
這篇文章主要介紹了python如何實現(xiàn)過濾敏感詞，幫助大家更好的理解和學習使用python，感興趣的朋友可以了解下2021-05-05
基于python3+OpenCV實現(xiàn)人臉和眼睛識別
這篇文章主要為大家詳細介紹了基于python3+OpenCV實現(xiàn)人臉和眼睛識別，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-09-09
python基于tkinter點擊按鈕實現(xiàn)圖片的切換
這篇文章主要介紹了python基于tkinter點擊按鈕實現(xiàn)圖片的切換，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-04-04
Pandas中DataFrame的分組/分割/合并的實現(xiàn)
這篇文章主要介紹了Pandas中DataFrame的分組/分割/合并的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-07-07