郑州网站设计专家,wordpress广告主题,微信公众号怎么创建账号,wordpress设置数字形链接报404文章目录 前言实现思路定义一个类然后开始手撸这个微型框架根据字符串获取到所定义的DTO类构建返回结果装饰器解析字符串#xff0c;获得变量SQL字符串拼接 使用装饰器 前言
在实际开发中#xff0c;根据业务拼接SQL所需要考虑的内容太多了。于是#xff0c;有没有一种办法… 文章目录 前言实现思路定义一个类然后开始手撸这个微型框架根据字符串获取到所定义的DTO类构建返回结果装饰器解析字符串获得变量SQL字符串拼接 使用装饰器 前言
在实际开发中根据业务拼接SQL所需要考虑的内容太多了。于是有没有一种办法可以像MyBatisPlus一样通过配置注解实现SQL注入呢
就像是
mybatis.select(select * from user where id #{id})
def get_user(id): ...那可就降低了好多工作量。 P.S.本文并不希望完全复现MyBatisPlus的所有功能能够基本配置SQL注解就基本能够完成大部分工作了。 实现思路
那我们这么考虑
首先我们需要定义一个类类中给一个或者多个装饰器我们先在类内定义一个字符串这个字符串能够配置到指定的DTO类用于存储结果我们针对装饰器中的SQL字符串进行解析解析到其中的变量个数与名称我们针对被装饰的函数进行解析与SQL变量进行匹配替换变量执行SQL
听起来并不难。我们一步步来。
定义一个类
首先定义
# dto/student.py
class Student:def __init__(self, name, age):self.name nameself.age age为了简化操作这个类就不放在任意位置了直接放在dto文件夹下后续导入这个类也就直接从dto文件夹中引入就不考虑做这个包名定位的接口了。
当然为了更方便后续的操作我们需要在dto文件夹中定义一个__init__.py文件用于对外暴露这个类
# dto/__init__.py
from dto.student import Student
__all__ [Student]最后呢我们为了方便这个类的序列化让他能够变成dict类型加一些魔法函数
# dto/student.py
class Student:def __init__(self, name, age):self.name nameself.age agedef __iter__(self):for key, value in self.__dict__.items():yield key, valuedef __getitem__(self, key):return getattr(self, key)def keys(self):return self.__dict__.keys()当然一个项目里面肯定不止这一个返回结果所以各位也可以这么操作
# dto/common.py
class CommonResult:def __init__(self): ...def __iter__(self):for key, value in self.__dict__.items():yield key, valuedef __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 nameself.age age至于实际业务中还有很多复杂的联立等操作需要新的类受限于篇幅就不展开了。如果能够把本篇看懂的话相信各位也没什么其他的困难了。 然后开始手撸这个微型框架
# db/common.py
from pydantic import BaseModel, Fieldclass DBManager(BaseModel):base_type: str Field(..., description数据库表名)link: str Field(..., description数据库连接地址)local_generator: Any Field(..., description实体类实例化解析生成器)def search(query_template): ...在这里呢我们定义了一个DBManager作为父类要求后面的子类必须有
str类型的base_type表示返回结果类的名称str类型的link表示数据库连接地址Any类型的local_generator表示实体类实例化解析生成器- 任意返回值的query方法用于执行SQL。 为什么一定要用BaseModel定义直接定义self.xxx不好吗 因为这样会看起来代码量很大逃 看着差不多。
根据字符串获取到所定义的DTO类
考虑到实际上我们所有的方法都需要特定到具体的位置所以这个方法还是直接写到DBManager类中这样子类就不需要再重写了。
# db/common.py
from pydantic import BaseModel, Fieldclass DBManager(BaseModel):base_type: str Field(..., description数据库表名)link: str Field(..., description数据库连接地址)local_generator: Any Field(..., description实体类实例化解析生成器)def search(query_template): ...def import_class_from_package(self, package_name, class_name):# 根据包名获得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 clselse:raise ImportError(f{class_name} not found in {package_name})这样子类就可以调用这个方法获得所需的类了。
构建返回结果
既然都已经能够动态导入类了那我把返回结果导入到Student中没问题吧
其中需要注意的是我这边采用的数据库驱动是sqlalchemy所以构造返回结果所需要的参数是sqlalchemy的Row类型。
同样的为了减少子类重写的代码量直接在父类给出来
# db/common.py
from pydantic import BaseModel, Field
from sqlalchemy.engine.row import Rowclass DBManager(BaseModel):base_type: str Field(..., description数据库表名)link: str Field(..., description数据库连接地址)local_generator: Any Field(..., description实体类实例化解析生成器)def search(query_template): ...# 为了方便看省略掉细节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装饰器
那么接下来就是重头戏了怎么定义这个装饰器。
我们先构建一个子类
# db/student.py
class StudentDBManager(DBManager):base_type: ClassVar[str] Studentlink: ClassVar[str] sqlite:///school.dblocal_generator: ClassVar[Any] None自定义PyMyBatisdef __init__(self):StudentDBManager.local_generator self.import_class_from_package(dto, self.base_type)在这里首先需要注意的是需要用ClassVar修饰将变量名定义为类内成员变量否则无法使用self.xxx访问。
其次我们利用base_type指定返回值对应的DTO类、link指定数据库连接地址local_generator指定实体类实例化解析生成器。
在这个类实例化的过程中我们还需要进一步构建local_generator也就是动态执行from xxx import xxx。
然后定义一个装饰器
def query(query_template: str):def decorator(func):wraps(func)def wrapper(*args, **kwargs):return func(*args, **kwargs)return wrapperreturn decorator这可以算得上是比较基础的模板了。至于之后怎么改管他呢先套公式。
在这里我们首先定义的装饰器是decorator没有参数其次再用query装饰器包装从而给无参的装饰器给一个参数从而接收一个SQL字符串参数。
好的我们再进一步。
解析字符串获得变量
首先当然是解析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 wrapperreturn decorator没啥问题。
接下来调用的时候我们需要检测是否完整给出了SQL字符串所需的参数。
我们考虑到如果但凡SQL中的参数有变化方法就会有变化因此每个SQL都有一个方法也太麻烦了。主要是这么多相似的方法起方法名太烦了
所以直接上反射获取 调用 的时侯传入的参数。
值得注意的是这里说的是 调用 的时候。因为Python中 定义 方法的时候可以使用**kargs传入多个参数但是如果反射直接获取到 定义 的参数将会只有一个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):# 获取函数的参数签名sig inspect.signature(func)bound_args sig.bind_partial(*args, **kwargs)bound_args.apply_defaults()# 提取传递的参数包括 **kwargs 中的参数provided_params set(bound_args.arguments.keys()) | set(kwargs.keys())# 检查缺失的参数missing_params required_params - provided_paramsif missing_params:raise ValueError(fMissing required parameters: {, .join(missing_params)})return func(*args, **kwargs)return wrapperreturn decorator这下应该就能够适配到所有的SQL情况了。
SQL字符串拼接
接下来就是直接替换值了。但是拼接真的就是对的吗我们不光是需要考虑不同的变量有着不同的植入格式同时也需要考虑到植入过程中可能的SQL注入问题。
所以我们就直接采用sqlalchemy的text函数对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):# 获取函数的参数签名sig inspect.signature(func)bound_args sig.bind_partial(*args, **kwargs)bound_args.apply_defaults()# 提取传递的参数包括 **kwargs 中的参数provided_params set(bound_args.arguments.keys()) | set(kwargs.keys())# 检查缺失的参数missing_params required_params - provided_paramsif missing_params:raise ValueError(fMissing required parameters: {, .join(missing_params)})# 构建 SQL 语句并考虑不同类型的数据格式sql_query text(query_template.replace(#{, :).replace(}, ))print(fExecuting SQL: {sql_query})return func(*args, **kwargs)return wrapperreturn decorator好了到这一步也就基本完成了。最后我们根据数据库存储数据的特点最后修整一下查询的格式细节就可以了
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):# 获取函数的参数签名sig inspect.signature(func)bound_args sig.bind_partial(*args, **kwargs)bound_args.apply_defaults()# 提取传递的参数包括 **kwargs 中的参数provided_params set(bound_args.arguments.keys()) | set(kwargs.keys())# 检查缺失的参数missing_params required_params - provided_paramsif missing_params:raise ValueError(fMissing required parameters: {, .join(missing_params)})# 构建 SQL 语句并考虑不同类型的数据格式sql_query text(query_template.replace(#{, :).replace(}, ))print(fExecuting 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_resultreturn wrapperreturn decorator就是这样我们就完成了这样一个装饰器。
使用装饰器
使用过程其实就可以类比Service中的调用了。而如果拿Python举例的话其实更像Flask的app.route。于是我们可以这么使用
sbd StudentDBManager()
sbd.query(SELECT * FROM student WHERE id #{id})
def find_student_by_id(**kargs): ...这也就实现了一个方法。
当然他也没那么智能。虽然写起来是这样但是依然相当于
sbd StudentDBManager()
sbd.query(SELECT * FROM student WHERE id #{id})
def find_student_by_id(id: str): ...只是说我们并不需要重复地去写驱动罢了。
