import pymysql
###-------------------属性描述器-------------------###
class Field: #描述器(类型判断) int varchar date enum float
def __init__(self,name,fieldname=None,pk=False,unique=False,default=None,nullable=True,index=False):
self.name = name
if fieldname is None: #字段名为空则将传入name当做字段名
self.fieldname = name
else:
self.fieldname =fieldname
self.pk = pk
self.unique = unique
self.default = default
self.nullable = nullable
self.index = index
def validate(self,value): #数据校验 分不同类型进行
raise NotImplementedError
def __get__(self, instance, owner): #instance是创建字段student的实例
if instance is None: #描述器的get返回的就是字段名--值的字典
return self
return instance.__dict__[self.name]
def __set__(self, instance, value): #设置字段的值 字段名--值
self.validate(value) #set之前先校验数据
instance.__dict__[self.name] = value #给对应属性名设置值
def __str__(self): #字段实例的本身是哪个 做区分
return "< {}类型的{}字段 >".format(self.__class__.__name__ , self.name)
__repr__ = __str__
class IntField(Field):
def __init__(self,name,fieldname=None,pk=False,unique=False,default=None,nullable=True,auto_increment=False,index=False):
self.auto_increment = auto_increment #int仅有
super().__init__(name,fieldname,pk,unique,default,nullable,index)
def validate(self,value): #auto_increment
if value is None:
if self.pk: #主键不可为None
raise TypeError(u"作为主键的%s 的属性%s 的值%s 异常" % (self.pk,self.name,value))
if not self.nullable: #如果当前为空 但数据库不允许为空时
raise TypeError(u"%s required" % self.name)
else:
if not isinstance(value,int): #判断是否为int型数据
raise TypeError(u"%s should be int" % self.name)
class StringField(Field): #字符串需要一个长度属性
def __init__(self,name,length=32,fieldname=None,pk=False,unique=False,default=None,nullable=True,index=False):
self.length = length
super().__init__(name,fieldname,pk,unique,default,nullable,index)
def validate(self,value):
if value is None:
if self.pk: #主键不可为None
raise (u"作为主键的%s 的属性%s 的值%s 异常" % (self.pk,self.name,value))
if not self.nullable: #如果当前为空 但数据库不允许为空时
raise TypeError(u"%s required"% self.name)
else:
if not isinstance(value,str): #判断是否为int型数据
raise TypeError(u"%s should be str"% self.name)
if len(value) > self.length:
raise ValueError(u"字段%s 长度超出范围,value=%s"% (self.name,value) )
###-------------------数据库操作会话 cursor操作的类-------------------###
class Session(): #这种模式会造成线程不安全 有一定风险 但Session应该不跨线程 全局使用
def __init__(self, conn):#:pymysql.connections.Connection):
self.conn = conn
self.cursor = None
print("初始化完成")
def execute(self, query,*args): #SQL执行通用方法 query参数接收SQL语句
#连接数据库 执行SQL语句 数据库的连接由外部传入，应是全局变量
if self.cursor is None: #判断保护 如果没有游标则新建一个
self.cursor = self.conn.cursor() #指令参数化 传入的是元组
print("已建立游标")
print("已存在游标")
self.cursor.execute(query,args)
def __enter__(self): #with Session实例操作时，单独获取一个游标
self.cursor = self.conn.cursor()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type: #有异常 回滚
self.conn.rollback()
else: #无异常 执行
self.conn.commit()
self.cursor.close()
###-------------------元类式反射 定义实体类的模板-------------------###
class ModelMeta(type): #元类只适合放共性相同的属性和方法
def __new__(cls, name:str, bases,attrs:dict): #attrs是 属性-值 的字典
#解决表名的问题 给了直接用 没给 就用name首字母大写作为表名
if attrs.get('__tablename__',None)is None: #获取表名 __tablename__
attrs['__tablename__'] = name.lower()
mapping = {} #映射字典
primarykey = [] #列表保存主键 主键不止一个！
for k,v in attrs.items(): #k属性名 v属性所对应的字段类型 保存在mapping字典中
if isinstance(v,Field):
mapping[k] = v #注意：这里在类构建时已经执行
if v.name is None:
v.name = k
if v.fieldname is None:
v.fieldname = v.name
if v.pk:
primarykey.append(v)
attrs['__mapping__'] = mapping #未定义时已经把字段属性放到映射字典中保存 作为__mapping__属性
attrs['primarykey'] = primarykey
return super().__new__(cls, name, bases,attrs)
"""
session.execute(sql , values)
将sql中的属性名(表名,字段名...)分离出来
values保存对应要添加的值
names保存属性名
"""
class Model(metaclass=ModelMeta): pass #基类来统一数据库操作
#实体类
class Engine:
def __init__(self,*args , **kwargs):
self.conn = pymysql.connect(*args , **kwargs)
def save(self,instance):
names = []
values = []
for k,v in instance.__mapping__.items(): #注意：当Model子类的实例调用方法时才调用
if isinstance(v,Field): #过滤出Field实例化的v
names.append(k) #k是属性名 访问v方式:
values.append(instance.__dict__[k]) #v是属性名对应的Field 访问v方式:__dict__[k]
#sql = "insert into student(id,name,age)values(%s,%s,%s)"
sql = "insert into {}({}) values ({})".format(instance.__tablename__ ,",".join(names), ",".join(['%s']*len(names))) #把values后面的参数拼成列表的形式[%s,%s,%s]
print(sql)
print(values)
session = Session(self.conn)
with session:
session.execute(sql, values)
###-------------------数据库连接和操作 实体类-------------------###
class Student(Model): #操作表student的记录
__tablename__ = "student"
id = IntField('id', 'id',True,nullable=False,auto_increment=True)#用描述器来指定属性名称(字段名) 按照int字段来创建 name fieldname pk null ...
name = StringField('name',length=64,nullable=False)
age = IntField('age') #默认age不做pk 默认就为False
def __init__(self,id,name,age):
self.tablename = ""
self.id = id #调用__set__方法 instance即为当前实例 字典的key为属性名称
self.name = name
self.age = age
# def __str__(self):
# return "Student(%s,%s,%s)" % (self.id,self.name,self.age)
if __name__ == '__main__':
s = Student(int(1),'tom',int(20))
engine = Engine(host = "10.0.0.6",port = 3306,user = "root",password='123',db = 'db1',charset='utf8')
engine.save(s)
报错： File "D:/Program Files/python code/数据分析/ORM关系映射操作MySQL/01ORM基本框架.py", line 100, in execute
self.cursor.execute(query,args)
File "D:\Program Files\Python\lib\site-packages\pymysql\cursors.py", line 146, in execute
query = self.mogrify(query, args)
File "D:\Program Files\Python\lib\site-packages\pymysql\cursors.py", line 125, in mogrify
query = query % self._escape_args(args, conn)
TypeError: not enough arguments for format string
