一 概念
1 概念
ORM ：对象关系映射，对象和关系之间的映射，使用面向对象的方式来操作数据库
关系对象模型和python对象模型之间的映射
tabel => class ,表映射类
row => object ,行映射为实例
column=> property ,字段映射属性
2 举例
表有login,字段为id int , username varchar, age int
映射到python为
#!/usr/bin/poython3.6
#conding:utf-8
class Login:
# 此处的INTX 是int类型的类
# VARY 是varchar类型的类
id=INTX()
username=VARY()
age=INTX()
# 最终得到
class Login:
def __init__(self):
self.id=1
self.username=‘admin‘
self.age=20
二 实现ORM 框架
1 字段类的实现
字段有name,字段名称为column，类型为type，是否主键pk，是否唯一键unique,是否索引index，是否可为空nullable，默认值default，是否自增等，这些都是字段的特征，所以字段可以使用类来描述。
字段类要提供对数据的校验功能，如声明字段类型为int类型，应该要判断数据是不是整数类形。
字段有多种类型，不同类型有差异，使用继承的方式实现。
字段现在定义为类属性，而这个类属性又适合使用类来描述，这就是描述器了。
1 定义基类，用于实现所有类的基础类型
#!/usr/bin/poython3.6
#conding:utf-8
class Field:
def __init__(self,name,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
self.name=name # 字段名称
if column is None: #列名称
self.column=self.name
else:
self.column=column
self.pk=pk # 主键
self.unique=unique #唯一
self.index=index #索引
self.nullable=nullable #是否为空
self.default=default # 默认是否为空
def validate(self,value): # 此处定义数据校验方式，每种不同类型的校验方式不同，因此应该在子类中分别实现
raise NotImplementedError #基类不实现此功能
def __get__(self, instance, owner): #此处用于定义描述器，此处当子类的类调用此属性时，会返回对应的值
# 此处的self表示父类的实例，instance表示子类的实例，owner表示子类的类
# pass
if instance is None: #此处为None表示子类未生成对应实例
return self # 此处的self表示实例自己
return instance.__dict__[self.name] # 返回实例对应的字段名称
def __set__(self, instance, value): #此处用于定义数据描述器，用于子类实例调用时使用，用于返回对应的结果
# instace 表示子类的实例，其相关的信息应该被存储于子类实例中，
self.validate(value)
instance.__dict__[self.name]=value
def __str__(self):
return "{} <{}>".format(self.__class__.__name__,self.name) # 此处返回被调用的类名和实例名称
__repr__=__str__
# 定义整数类型的类型属性
class IntField(Field): #多了自增属性。
def __init__(self,name,column=None,pk=False,unique=False,index=False,nullable=True,default=None,auto_increment=True):
self.auto_increment=auto_increment
super().__init__(name,column,pk,unique,index,nullable,default)
def validate(self,value):
if value is None:
if self.pk: # 主键不能为空，因此此处会报错
raise TypeError("{}:{}".format(self.name,value))
if not self.nullable: # 当定义了非空时，上述的值为空，则报错
raise TypeError
else:
if not isinstance(value,int): #若数据的类型为非int，则报错
raise TypeError("{} is not int, It‘s {}".format(self.name,type(value)))
# 定义字符串的类型属性
class StringField(Field): #定义字符串属性类
# 增加了字符串长度的定义
def __init__(self,length=32,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
super().__init__(name,column,pk,unique,index,nullable,default) #此处的属性可以继承父类的属性
self.length=length #此处用于定义字符串类型的长度
def validate(self,value): #此处用于定义各自的属性检查,对数据进行属性检查
if value is None: # 此处的None对应数据库的null
if self.pk: # 如果数据是None，而其定义了主键，则会报错，因为主键必须不能是Null，主键非空且唯一
raise TypeError("{} is pk,not None".format(self.name))
if not self.nullable: # 如果其是None,而定义的是非null，则会报错
raise TypeError("{} is not null".format(self.name))
else:
if not isinstance(value,str):
raise TypeError("{} should be string".format(self.name))
if len(value) > self.length: #真实的值大于规定的值，则会报错
raise ValueError("{} is to long value={}".format(self.name,value))2 Login 类的实现
# 具体类的实现
class Login:
id=IntField(‘id‘,‘id‘,pk=True,nullable=False,auto_increment=True) # 此种调用方式会启动get方法的调用，从而返回实例自己
name=StringField(length=64,name=‘username‘,nullable=False)
age=IntField(‘age‘)
def __init__(self,id,nane,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__Login 类的操作
Login类的操作对应表的CRUD操作，及增删改查，如果使用pymysql，应该使用cursor对象的execute方法，增加数据，修改数据定义为save方法，为Login类增加此方法，数据库的链接要求从外面传入具体实现如下
# 具体类的实现
# 具体类的实现
class Login:
id=IntField(‘id‘,‘id‘,pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,name=‘username‘,nullable=False)
age=IntField(‘age‘)
def __init__(self,id,nane,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
def save(self,conn:pymysql.connections.Connection):
sql="insert into login(id,bane,age) values(%s,%s,%s)"
with conn as cursor:
cursor.execute(sql,(self.id,self.name,self.age))3 session类的实现
每一次数据库操作都是在一个会话中完成的，将cursor的操作封装在会话中
class Session: #此处用以封装链接，可在此处增加上下文支持
def __init__(self,conn:pymysql.connections.Connection):
self.conn=conn
self.cursor=None
def execute(self,query,*args):
if self.cursor is None:
self.cursor=self.conn.cursor()
self.cursor.execute(query,args)
def __enter__(self): # 此处实现方式和
return self.conn.cursor()
# self.cursor=self.conn.cursor()
# return self 如此写，这个session必须是一个线程级别的，如果用进程，则直接覆盖cursor
# #因为线程是顺序执行的，都用新的cursor（）当查询数据时，数据找不到了，因为cursor变了。本session是在线程内执行，不能夸线程执行
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
if exc_type: # 此处用于定义是否出错，若出错，则直接返回
self.conn.rollback()
else:
self.conn.commit()
class Login:
id=IntField(‘id‘,‘id‘,pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,name=‘username‘,nullable=False)
age=IntField(‘age‘)
def __init__(self,id,name,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
def save(self,session:Session):
sql="insert into login(id,name,age) values(%s,%s,%s)"
with session as cursor: # 此处直接调用enter返回cursor游标，此处的最后会关闭游标，但不会关闭链接
with cursor: # 此处使用游标的属性进行处理
cursor.execute(sql,(self.id,self.name,self.age))4 Model 类的实现
Login 这样的类，如果多建立几个，其都是一个样子，每一个这样的类，得定义一个名称对应不同的表，都需要先定义好类，再定义__init__初始化值，而这些值刚好是定义好的类属性，操作也是一样的。CRID
设计，定义一个Model类，增加一个__table__类属性来保存不同的表名称
class Model:
def save(self,session:Session=None):
names=[]
values=[]
for k,v in self.__class__.__dict__.items(): # 此处用于获取实例的名称和其对应的值
# print (‘for‘,k,‘---‘,v)
if isinstance(v,Field): # 此处若属于基类
if k in self.__dict__.keys(): # 此处的字段符合
names.append(k)
values.append(v)
# __table__ # 此处在子类中添加
query="insert into {} ({}) values ({})".format(self.__table__,",".join(names),",".join(["%s"]*len(values))) # 此处是匹配对应的sql
print (query)
print (values)
class Login(Model):
id=IntField(‘id‘,‘id‘,pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,name=‘name‘,nullable=False)
age=IntField(‘age‘)
__table__=‘Login‘
def __init__(self,id,name,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
5 使用元类改造Model
编写一个元类ModelMeta
以它作为元类的类，都可以获得一个类属性
如果没有定义_table_，就自动加上这个属性，值为类名
可以遍历类属性，找出定义的字段类，建立一张映射表mapping
找出主键字段primarykey
#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
class Field:
def __init__(self,name,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
self.name=name # 字段名称
if column is None: #列名称
self.column=self.name
else:
self.column=column
self.pk=pk # 主键
self.unique=unique #唯一
self.index=index #索引
self.nullable=nullable #是否为空
self.default=default # 默认是否为空
def validate(self,value): # 此处定义数据校验方式，每种不同类型的校验方式不同，因此应该在子类中分别实现
raise NotImplementedError #基类不实现此功能
def __get__(self, instance, owner): #此处用于定义描述器，此处当子类的类调用此属性时，会返回对应的值
# 此处的self表示父类的实例，instance表示子类的实例，owner表示子类的类
# pass
if instance is None: #此处为None表示子类未生成对应实例
return self # 此处的self表示实例自己
return instance.__dict__[self.name] # 返回实例对应的字段名称
def __set__(self, instance, value): #此处用于定义数据描述器，用于子类实例调用时使用，用于返回对应的结果
# instace 表示子类的实例，其相关的信息应该被存储于子类实例中，
self.validate(value)
instance.__dict__[self.name]=value #此处的name是字段名，value是子类的self,对应的属性的值，及真实的数据
def __str__(self):
return "{} <{}>".format(self.__class__.__name__,self.name) # 此处返回被调用的类名和实例名称
__repr__=__str__
# 定义整数类型的类型属性
class IntField(Field): #多了自增属性。
def __init__(self,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None,auto_increment=True):
self.auto_increment=auto_increment
super().__init__(name,column,pk,unique,index,nullable,default)
def validate(self,value):
if value is None:
if self.pk: # 主键不能为空，因此此处会报错
raise TypeError("{}:{}".format(self.name,value))
if not self.nullable: # 当定义了非空时，上述的值为空，则报错
raise TypeError
else:
if not isinstance(value,int): #若数据的类型为非int，则报错
raise TypeError("{} is not int, It‘s {}".format(self.name,type(value)))
# 定义字符串的类型属性
class StringField(Field): #定义字符串属性类
# 增加了字符串长度的定义
def __init__(self,length=32,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
super().__init__(name,column,pk,unique,index,nullable,default) #此处的属性可以继承父类的属性
self.length=length #此处用于定义字符串类型的长度
def validate(self,value): #此处用于定义各自的属性检查,对数据进行属性检查
if value is None: # 此处的None对应数据库的null
if self.pk: # 如果数据是None，而其定义了主键，则会报错，因为主键必须不能是Null，主键非空且唯一
raise TypeError("{} is pk,not None".format(self.name))
if not self.nullable: # 如果其是None,而定义的是非null，则会报错
raise TypeError("{} is not null".format(self.name))
else:
if not isinstance(value,str):
raise TypeError("{} should be string".format(self.name))
if len(value) > self.length: #真实的值大于规定的值，则会报错
raise ValueError("{} is to long value={}".format(self.name,value))
# 具体类的实现
class Session: #此处用以封装链接，可在此处增加上下文支持
def __init__(self,conn:pymysql.connections.Connection):
self.conn=conn
self.cursor=None
def execute(self,query,*args):
if self.cursor is None:
self.cursor=self.conn.cursor()
self.cursor.execute(query,args)
def __enter__(self): # 此处实现方式和
return self.conn.cursor()
# self.cursor=self.conn.cursor()
# return self 如此写，这个session必须是一个线程级别的，如果用进程，则直接覆盖cursor
# #因为线程是顺序执行的，都用新的cursor（）当查询数据时，数据找不到了，因为cursor变了。本session是在线程内执行，不能夸线程执行
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
if exc_type: # 此处用于定义是否出错，若出错，则直接返回
self.conn.rollback()
else:
self.conn.commit()
class ModeMetd(type): # 子类构建的时候实现的
def __new__(cls,name,bases,attrs:dict): # 此处是在类的定义中进行调用的，
# 此处的name表示被调用子类的类名，而对应的字典attrs则表示子类的属性字典
# name 类名，attrs类属性字典
if ‘__table__‘ not in attrs.keys():
attrs[‘__table__‘]=name #默认添加表名称为类名称
mapping={} # 方便后面查询属性名和字段实例
primarykey=[] # 多个主键的情况下使用，如果使用一个变量名，则导致后面覆盖前面
for k,v in attrs.items(): # k代表的是列名称，v表示的是子类的类型
if isinstance(v,Field): # 此处判断是否继承与父类
mapping[k]=v
if v.name is None: # 此处用于处理子类的属性的字典的列名称处理问题
v.name=k # 如果是，则将类.name
if v.column is None:
v.column=v.name # 没有给字段名,则使用类对应的列名称
if v.pk:
primarykey.append(v)
# 增加属性
attrs[‘__mapping__‘]=mapping
attrs[‘__primarykey__‘]=primarykey
return super().__new__(cls,name,bases,attrs)
class Model(metaclass=ModeMetd): # 实体类的调用时实现，子类的实例调用时实现的
def save(self,session:Session=None):
names=[]
values=[]
for k,v in self.__class__.__dict__.items(): # 此处用于获取实例的名称和其对应的值
# print (‘for‘,k,‘---‘,v)
if isinstance(v,Field): # 此处若属于基类
if k in self.__dict__.keys(): # 此处的字段符合
names.append(k)
values.append(self.__dict__[k]) # v是一个Field类型的实例，是子类和父类的实例
print (self.__dict__[k]) # 此处是实例。实例中的数字
# __table__ # 此处在子类中添加
query="insert into {} ({}) values ({})".format(self.__table__,",".join(names),",".join(["%s"]*len(values))) # 此处是匹配对应的sql
print (query)
# with session:
# session.execute(query,values)
class Login(Model):
id=IntField(pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,name=‘name‘,nullable=False)
age=IntField()
__table__=‘Login‘
def __init__(self,id,name,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
l=Login(1,‘admin‘,20)
l.save(None)
if __name__ == "__main__":
pass结果如下
6 引擎类
实体类没有提供数据库连接，当然也不应该提供，实体类就应该只完成表和类的映射。
提供一个数据库的包装类
1 负责数据库连接
2 负责CRUD操作，取代实体类的CRUD方法
#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
class Field:
def __init__(self,name,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
self.name=name # 字段名称
if column is None: #列名称
self.column=self.name
else:
self.column=column
self.pk=pk # 主键
self.unique=unique #唯一
self.index=index #索引
self.nullable=nullable #是否为空
self.default=default # 默认是否为空
def validate(self,value): # 此处定义数据校验方式，每种不同类型的校验方式不同，因此应该在子类中分别实现
raise NotImplementedError #基类不实现此功能
def __get__(self, instance, owner): #此处用于定义描述器，此处当子类的类调用此属性时，会返回对应的值
# 此处的self表示父类的实例，instance表示子类的实例，owner表示子类的类
# pass
if instance is None: #此处为None表示子类未生成对应实例
return self # 此处的self表示实例自己
return instance.__dict__[self.name] # 返回实例对应的字段名称
def __set__(self, instance, value): #此处用于定义数据描述器，用于子类实例调用时使用，用于返回对应的结果
# instace 表示子类的实例，其相关的信息应该被存储于子类实例中，
self.validate(value)
instance.__dict__[self.name]=value #此处的name是字段名，value是子类的self,对应的属性的值，及真实的数据
def __str__(self):
return "{} <{}>".format(self.__class__.__name__,self.name) # 此处返回被调用的类名和实例名称
__repr__=__str__
# # 定义整数类型的类型属性
class IntField(Field): #多了自增属性。
def __init__(self,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None,auto_increment=True):
self.auto_increment=auto_increment
super().__init__(name,column,pk,unique,index,nullable,default)
def validate(self,value):
if value is None:
if self.pk: # 主键不能为空，因此此处会报错
raise TypeError("{}:{}".format(self.name,value))
if not self.nullable: # 当定义了非空时，上述的值为空，则报错
raise TypeError
else:
if not isinstance(value,int): #若数据的类型为非int，则报错
raise TypeError("{} is not int, It‘s {}".format(self.name,type(value)))
# 定义字符串的类型属性
class StringField(Field): #定义字符串属性类
# 增加了字符串长度的定义
def __init__(self,length=32,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
super().__init__(name,column,pk,unique,index,nullable,default) #此处的属性可以继承父类的属性
self.length=length #此处用于定义字符串类型的长度
def validate(self,value): #此处用于定义各自的属性检查,对数据进行属性检查
if value is None: # 此处的None对应数据库的null
if self.pk: # 如果数据是None，而其定义了主键，则会报错，因为主键必须不能是Null，主键非空且唯一
raise TypeError("{} is pk,not None".format(self.name))
if not self.nullable: # 如果其是None,而定义的是非null，则会报错
raise TypeError("{} is not null".format(self.name))
else:
if not isinstance(value,str):
raise TypeError("{} should be string".format(self.name))
if len(value) > self.length: #真实的值大于规定的值，则会报错
raise ValueError("{} is to long value={}".format(self.name,value))
# 具体类的实现
class Session: #此处用以封装链接，可在此处增加上下文支持
def __init__(self,conn:pymysql.connections.Connection):
self.conn=conn
self.cursor=None
def execute(self,query,*args):
if self.cursor is None:
self.cursor=self.conn.cursor()
self.cursor.execute(query,args)
def __enter__(self): # 此处实现方式和
return self.conn.cursor()
# self.cursor=self.conn.cursor()
# return self 如此写，这个session必须是一个线程级别的，如果用进程，则直接覆盖cursor
# #因为线程是顺序执行的，都用新的cursor（）当查询数据时，数据找不到了，因为cursor变了。本session是在线程内执行，不能夸线程执行
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
if exc_type: # 此处用于定义是否出错，若出错，则直接返回
self.conn.rollback()
else:
self.conn.commit()
class ModeMetd(type): # 子类构建的时候实现的
def __new__(cls,name,bases,attrs:dict): # 此处是在类的定义中进行调用的，
# 此处的name表示被调用子类的类名，而对应的字典attrs则表示子类的属性字典
# name 类名，attrs类属性字典
if ‘__table__‘ not in attrs.keys():
attrs[‘__table__‘]=name #默认添加表名称为类名称
mapping={} # 方便后面查询属性名和字段实例
primarykey=[] # 多个主键的情况下使用，如果使用一个变量名，则导致后面覆盖前面
for k,v in attrs.items(): # k代表的是列名称，v表示的是子类的类型
if isinstance(v,Field): # 此处判断是否继承与父类
mapping[k]=v
if v.name is None:
v.name=k # 如果是，则将类.name
if v.column is None:
v.column=v.name # 没有给字段名,则使用类对应的列名称
if v.pk:
primarykey.append(v)
# 增加属性
attrs[‘__mapping__‘]=mapping
attrs[‘__primarykey__‘]=primarykey
return super().__new__(cls,name,bases,attrs)
class Model(metaclass=ModeMetd): # 实体类的调用时实现，子类的实例调用时实现的
pass
class Login(Model):
id=IntField(pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,nullable=False)
age=IntField()
__table__=‘Login‘
def __init__(self,id,name,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
class Engine:
def __init__(self,*args,**kwargs):
self.conn=pymysql.Connect(*args,**kwargs)
def save(self, instance:Login):
names = []
values = []
for k, v in instance.__mapping__.items(): # 此处用于获取实例的名称和其对应的值
# print (‘for‘,k,‘---‘,v)
if isinstance(v, Field): # 此处若属于基类
if k in instance.__dict__.keys(): # 此处的字段符合
names.append(k)
values.append(instance.__dict__[k]) # v是一个Field类型的实例，是子类和父类的实例
print(instance.__dict__[k]) # 此处是实例。实例中的数字
# __table__ # 此处在子类中添加
query = "insert into {} ({}) values ({})".format(instance.__table__, ",".join(names),
",".join(["%s"] * len(values))) # 此处是匹配对应的sql
print(query)
print (values)
l=Login(1,‘admin‘,20)
e=Engine(‘192.168.1.120‘,‘root‘,‘666666‘,‘test‘)
e.save(l)
基础结果如下
#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
class Field:
def __init__(self,name,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
self.name=name # 字段名称
if column is None: #列名称
self.column=self.name
else:
self.column=column
self.pk=pk # 主键
self.unique=unique #唯一
self.index=index #索引
self.nullable=nullable #是否为空
self.default=default # 默认是否为空
def validate(self,value): # 此处定义数据校验方式，每种不同类型的校验方式不同，因此应该在子类中分别实现
raise NotImplementedError #基类不实现此功能
def __get__(self, instance, owner): #此处用于定义描述器，此处当子类的类调用此属性时，会返回对应的值
# 此处的self表示父类的实例，instance表示子类的实例，owner表示子类的类
# pass
if instance is None: #此处为None表示子类未生成对应实例
return self # 此处的self表示实例自己
return instance.__dict__[self.name] # 返回实例对应的字段名称
def __set__(self, instance, value): #此处用于定义数据描述器，用于子类实例调用时使用，用于返回对应的结果
# instace 表示子类的实例，其相关的信息应该被存储于子类实例中，
self.validate(value)
instance.__dict__[self.name]=value #此处的name是字段名，value是子类的self,对应的属性的值，及真实的数据
def __str__(self):
return "{} <{}>".format(self.__class__.__name__,self.name) # 此处返回被调用的类名和实例名称
__repr__=__str__
# 定义整数类型的类型属性
class IntField(Field): #多了自增属性。
def __init__(self,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None,auto_increment=True):
self.auto_increment=auto_increment
super().__init__(name,column,pk,unique,index,nullable,default)
def validate(self,value):
if value is None:
if self.pk: # 主键不能为空，因此此处会报错
raise TypeError("{}:{}".format(self.name,value))
if not self.nullable: # 当定义了非空时，上述的值为空，则报错
raise TypeError
else:
if not isinstance(value,int): #若数据的类型为非int，则报错
raise TypeError("{} is not int, It‘s {}".format(self.name,type(value)))
# 定义字符串的类型属性
class StringField(Field): #定义字符串属性类
# 增加了字符串长度的定义
def __init__(self,length=32,name=None,column=None,pk=False,unique=False,index=False,nullable=True,default=None):
super().__init__(name,column,pk,unique,index,nullable,default) #此处的属性可以继承父类的属性
self.length=length #此处用于定义字符串类型的长度
def validate(self,value): #此处用于定义各自的属性检查,对数据进行属性检查
if value is None: # 此处的None对应数据库的null
if self.pk: # 如果数据是None，而其定义了主键，则会报错，因为主键必须不能是Null，主键非空且唯一
raise TypeError("{} is pk,not None".format(self.name))
if not self.nullable: # 如果其是None,而定义的是非null，则会报错
raise TypeError("{} is not null".format(self.name))
else:
if not isinstance(value,str):
raise TypeError("{} should be string".format(self.name))
if len(value) > self.length: #真实的值大于规定的值，则会报错
raise ValueError("{} is to long value={}".format(self.name,value))
# 具体类的实现
class Session: #此处用以封装链接，可在此处增加上下文支持
def __init__(self,conn:pymysql.connections.Connection):
self.conn=conn
self.cursor=None
def execute(self,query,*args):
if self.cursor is None:
self.cursor=self.conn.cursor()
self.cursor.execute(query,*args)
def __enter__(self): # 此处实现方式和
return self.conn.cursor()
# self.cursor=self.conn.cursor()
# return self 如此写，这个session必须是一个线程级别的，如果用进程，则直接覆盖cursor
# 因为线程是顺序执行的，都用新的cursor（）当查询数据时，数据找不到了，因为cursor变了。本session是在线程内执行，不能夸线程执行
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
if exc_type: # 此处用于定义是否出错，若出错，则直接返回
self.conn.rollback()
else:
self.conn.commit()
class ModeMetd(type): # 子类构建的时候实现的
def __new__(cls,name,bases,attrs:dict): # 此处是在类的定义中进行调用的，
# 此处的name表示被调用子类的类名，而对应的字典attrs则表示子类的属性字典
# name 类名，attrs类属性字典
if ‘__table__‘ not in attrs.keys():
attrs[‘__table__‘]=name #默认添加表名称为类名称
mapping={} # 方便后面查询属性名和字段实例
primarykey=[] # 多个主键的情况下使用，如果使用一个变量名，则导致后面覆盖前面
for k,v in attrs.items(): # k代表的是列名称，v表示的是子类的类型
if isinstance(v,Field): # 此处判断是否继承与父类
mapping[k]=v
if v.name is None:
v.name=k # 如果是，则将类.name
if v.column is None:
v.column=v.name # 没有给字段名,则使用类对应的列名称
if v.pk:
primarykey.append(v)
# 增加属性
attrs[‘__mapping__‘]=mapping
attrs[‘__primarykey__‘]=primarykey
return super().__new__(cls,name,bases,attrs)
class Model(metaclass=ModeMetd): # 实体类的调用时实现，子类的实例调用时实现的
pass
class Login(Model):
id=IntField(pk=True,nullable=False,auto_increment=True)
name=StringField(length=64,nullable=False)
age=IntField()
__table__=‘login‘
def __init__(self,id,name,age):
self.id=id
self.name=name
self.age=age
def __str__(self):
return "Loin({},{},{})".format(self.id,self.name,self.age)
__repr__=__str__
class Engine:
def __init__(self,*args,**kwargs):
self.conn=pymysql.connections.Connection(*args,**kwargs)
def save(self, instance:Login):
names = []
values = []
for k, v in instance.__mapping__.items(): # 此处用于获取实例的名称和其对应的值
# print (‘for‘,k,‘---‘,v)
if isinstance(v, Field): # 此处若属于基类
if k in instance.__dict__.keys(): # 此处的字段符合
names.append(k)
values.append(instance.__dict__[k]) # v是一个Field类型的实例，是子类和父类的实例
# __table__ # 此处在子类中添加
query = "insert into {}({}) values({})".format(instance.__table__, ",".join(names),
",".join(["%s"] * len(values))) # 此处是匹配对应的sql
S=Session(self.conn)
with S:
S.execute(query,values)
e=Engine(‘192.168.1.200‘,‘root‘,‘[email protected]‘,‘test‘)
for i in range(10):
e.save(Login(i,‘admin‘+str(i),20+i))结果如下
原文地址：https://blog.51cto.com/11233559/2440408
时间： 2024-10-02 05:02:08