利用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)思路
那我們這么考慮:
- 首先,我們需要定義一個(gè)類,類中給一個(gè)或者多個(gè)裝飾器;
- 我們先在類內(nèi)定義一個(gè)字符串,這個(gè)字符串能夠配置到指定的
DTO
類,用于存儲(chǔ)結(jié)果; - 我們針對(duì)裝飾器中的
SQL
字符串進(jìn)行解析,解析到其中的變量個(gè)數(shù)與名稱; - 我們針對(duì)被裝飾的函數(shù)進(jìn)行解析,與
SQL
變量進(jìn)行匹配; - 替換變量;
- 執(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ù)是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="實(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
注入問題。
所以,我們就直接采用sqlalchemy
的text
函數(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í)更像Flask
的app.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)代碼示例,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-11-11Python標(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),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-07-07如何將你的應(yīng)用遷移到Python3的三個(gè)步驟
這篇文章主要介紹了如何將你的應(yīng)用遷移到Python3的三個(gè)步驟,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12python腳本爬取字體文件的實(shí)現(xiàn)方法
這篇文章主要給大家介紹了利用python腳本爬取字體文件的實(shí)現(xiàn)方法,文中分享了爬取兩個(gè)不同網(wǎng)站的示例代碼,相信對(duì)大家具有一定的參考價(jià)值,需要的朋友們下面來一起看看吧。2017-04-04Python3 把一個(gè)列表按指定數(shù)目分成多個(gè)列表的方式
今天小編就為大家分享一篇Python3 把一個(gè)列表按指定數(shù)目分成多個(gè)列表的方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-12-12xadmin使用formfield_for_dbfield函數(shù)過濾下拉表單實(shí)例
這篇文章主要介紹了xadmin使用formfield_for_dbfield函數(shù)過濾下拉表單實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2020-04-04python實(shí)現(xiàn)簡(jiǎn)單遺傳算法
這篇文章主要介紹了python如何實(shí)現(xiàn)簡(jiǎn)單遺傳算法,幫助大家更好的利用python進(jìn)行數(shù)據(jù)分析,感興趣的朋友可以了解下2020-09-09