元编程
type类
class type(object): """ type(object_or_name, bases, dict) type(object) -> the object's type #返回对象的类型,如type(10) type(name, bases, dict) -> a new type #返回一个新的类型 """ pass
def show(self): print(self.__dict__) def __init__(self): self.x =100 #借助type构造任何类,用代码来生成代码,这就是元编程 C = type('myclass',(object,),{'b':100,'show':show,'__init__':__init__}) print(C) #<class '__main__.B'> print(type(C)) #type print(C.__name__) #myclass print(C.__dict__) #{'b': 100, 'show': <function show at 0x0000000001D21E18>, ...} print(C.mro()) #[<class '__main__.B'>, <class 'object'>]
#利用继承type的方法,可以认为ModelMeta就是元类,它可以创造出类 class ModelMeta(type): #元类 def __new__(cls, *args, **kwargs): print(cls) print(args) print(kwargs,"----") return super().__new__(cls,*args,**kwargs) #注意下面两种构造方式的区别 class A(ModelMeta): pass class B(metaclass=ModelMeta): pass #相当于 B=type('B',(object,),{}) class C(B): pass print(A) #<class '__main__.A'> print(type(A)) #<class 'type'> print(A.__mro__) #(<class '__main__.A'>, <class '__main__.ModelMeta'>, <class 'type'>, <class 'object'>) print(B) #<class '__main__.B'> print(type(B)) #<class '__main__.ModelMeta'> print(B.__mro__) #(<class '__main__.B'>, <class 'object'>) print(C) #<class '__main__.C'> print(type(C)) #<class '__main__.ModelMeta'> print(C.__mro__) #(<class '__main__.C'>, <class '__main__.B'>, <class 'object'>)
#第一种 使用metaclass关键字参数指定元类 class A(metaclass=ModelMeta): id =100 def __init__(self): print('A.init') #第二种 B继承自A后,依然是从ModelMeta的类型 class B(A): def __init__(self): print('B.init') #第三种 元类可以使用下面的方式创建新的类 D = ModelMeta('D',(),{}) print(D.__mro__) #(<class '__main__.D'>, <class 'object'>) #C、E是type的实例 class C:pass #C = type('C',(),{}) E = type('E',(),{}) class F(ModelMeta):pass print(type(A)) #<class '__main__.ModelMeta'> print(type(B)) #<class '__main__.ModelMeta'> print(type(D)) #<class '__main__.ModelMeta'> print(type(C)) #<class 'type'> print(type(E)) #<class 'type'> print(type(F)) #<class 'type'> print(A.__mro__) #(<class '__main__.A'>, <class 'object'>) print(B.__mro__) #(<class '__main__.B'>, <class '__main__.A'>, <class 'object'>) print(D.__mro__) #(<class '__main__.D'>, <class 'object'>) print(C.__mro__) #(<class '__main__.C'>, <class 'object'>) print(E.__mro__) #(<class '__main__.E'>, <class 'object'>) print(F.__mro__) #(<class '__main__.F'>, <class '__main__.ModelMeta'>, <class 'type'>, <class 'object'>)
class ModelMeta(type): #继承自type def __new__(cls, name,bases,dict): print(cls) print(name) print(bases) print(dict) return super().__new__(cls,name,bases,dict)
Python之 __new__()与__init__()方法详解 参考:https://www.cnblogs.com/ifantastic/p/3175735.html
元类的应用
# ORM对象关系映射 class Field: def __init__(self,fieldname=None,pk=False,nullable=True): self.fieldname = fieldname self.pk = pk self.nullable = nullable def __repr__(self): return "<Field {}>".format(self.fieldname) class ModelMeta(type): def __new__(cls, name,bases,attrs:dict): print(cls) print(name) print(bases) print(attrs,'-------') if '__tablebname__' not in attrs.keys(): attrs['__tablename__'] = name primarykeys = [] for k,v in attrs.items(): if isinstance(v,Field): if v.fieldname is None: v.fieldname = k #没有名字则使用属性名 if v.pk: primarykeys.append(v) attrs['__primarykeys__'] = primarykeys return super().__new__(cls,name,bases,attrs) class ModelBase(metaclass=ModelMeta): #从ModelBase继承的类的类型都是ModelMeta pass class Student(ModelBase): id = Field(pk=True,nullable=False) name = Field('username',nullable=False) age = Field()
元编程总结
ORM
#关系模型和Python对象之间的映射 table => class 表映射为类 row => object 行映射为实例 column => property 字段映射为属性
class Student: id = ?某字段类型 name = ?某字段类型 age = ?某字段类型 最终得到实例 class Student: def __init__(self): self.id = ? self.name = ? self.age = ?
SQLAlchemy
安装
$pip install sqlalchemy
文档
#官方文档 http://docs.sqlalchemy.org/enlatest/ #查看版本 import sqlalchemy print(sqlalchemy.__version__)
开发
创建连接
#pymysql的连接 #mysql+pymysql://<username>:<password>@<host>:<port>/<dbname> from sqlalchemy import create_engine connstr = "{}://{}:{}@{}:{}/{}".format( 'mysql+pymysql', 'wayne','wayne', '192.168.10.166',3306, 'school' ) engine = create_engine(connstr,echo=True)
Declare a Mapping创建映射
创建基类
from sqlalchemy.ext.declarative import declarative_base #创建基类,便于实体类继承 Base = declarative_base()
创建实体类
student表
CREATE TABLE student ( id INTEGER NOT NULL AUTO_INCREMENT, name VARCHAR(64), age INTEGER, PRIMARY KEY (id) )
class Student(Base): __tablename__ = "student" #指定表名 #制定对应属性字段 id = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(64),nullable=True) age = Column(Integer) #第一参数是字段名,如果和属性名不一致,一定要指定 #age = Column('age',Interger) def __repr__(self): return "<{}> {} {} {}".format(self.__class__.__name__,self.id,self.name,self.age)
实例化
s = Student(name='tom') s.age = 20
创建表
#删除继承自Base的所有表 Base.metadata.create_all(bind=engine) #创建继承自Base的所有表 Base.metadata.drop_all(bind=engine)
创建会话
#创建会话session Session = sessionmaker(bind=engine) #返回类 session = Session() #实例化
CRUD操作
增
#添加语句 session.add(s1) try: session.commit() except Exception as e: print(e) session.rollback()
下面这种情况会出现主键冲突,当前数据库中数据表如下
#... #实例化 s1 = Student(name='tom') s1.age = 20 s2 = Student(name='jerry') s2.id = '4' #添加语句 session.add(s1) session.commit() try: session.add_all([s2]) session.commit() except Exception as e: print(e) session.rollback()
简单查询
#查询语句 students = session.query(Student)#无条件 for student in students: print(student) print(type(student)) student = session.query(Student).get(2) #通过主键查询 print(student)
改
#修改 student = session.query(Student).get(2) student.name = 'sam' student.age = 30 session.add(student) session.commit()
删除
先看下数据库,表中有
#运行如下程序 #... try: student = Student(id=2,name="tom",age=20) session.delete(student) session.commit() except Exception as e: print(e) #产生异常 # Instance '<Student at 0x49e3978>' is not persisted
状态**
import sqlalchemy from sqlalchemy import create_engine,Column,Integer,String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker connstr = "{}://{}:{}@{}:{}/{}".format( "mysql+pymysql","wayne","wayne", "192.168.10.166",3306,"school" ) engine = create_engine(connstr,echo=True) Base = declarative_base() #创建实体类 class Student(Base): __tablename__ = "student" id = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(64),nullable=True) age = Column(Integer) def __repr__(self): return "{} id={} name={} age={}".format( self.__class__.__name__,self.id,self.name,self.age ) Session = sessionmaker(bind=engine) session = Session() def getstate(entity,i): insp = sqlalchemy.inspect(entity) state = "sessionid={},attached={} transient={},persistent={} pending={},deleted={},detanched={}".format( insp.session_id, insp._attached, insp.transient, insp.persistent, insp.pending, insp.deleted, insp.detached ) print(i,state) print(insp.key) print("="*30) student = session.query(Student).get(2) getstate(student,1) #persistent try: student = Student(id=5,name='tom',age=30) getstate(student,2)#transinet student = Student(name="sammy",age=30) getstate(student,3)#transient session.add(student) getstate(student,4)#pending session.commit() getstate(student,5)#persistent except Exception as e: session.rollback() print(e) print("~~~~~~~")
student = session.query(Student).get(2) getstate(student,10) #persistent try: session.delete(student) #删除的前提是persistent,否则抛异常 getstate(student,11)#persistent session.flush() getstate(student,12)#deleted session.commit() getstate(student,13)#detached except Exception as e: session.rollback() print(e) print("~~~~~~~")
复杂查询
from sqlalchemy import create_engine,Column,Integer,String,Date,Enum from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import enum connstr = "{}://{}:{}@{}:{}/{}".format( "mysql+pymysql","wayne","wayne", "192.168.10.166",3306,"test" ) engine = create_engine(connstr,echo=True) Base = declarative_base(bind=engine) Session = sessionmaker() session = Session() #枚举类 class MyEnum(enum.Enum): M = 'M' F = 'F' #实体类 class Employees(Base): __tablename__ = "employees" emp_no = Column(Integer,primary_key=True) birth_date = Column(Date,nullable=False) first_name = Column(String(14),nullable=False) last_name = Column(String(16),nullable=False) gender = Column(Enum(MyEnum),nullable=False) hire_date = Column(Date,nullable=False) def __repr__(self): return "{} no={} name={} {} gender={}".format( self.__class__.__name__,self.emp_no,self.first_name,self.last_name,self.gender.value ) #打印函数 def show(emps): for x in emps: print(x) print('~~~~~~~~~~~~',end=' ') #简单查询条件 # emps = session.query(Employees).filter(Employees.emp_no>10015) # show(emps) #与或非 from sqlalchemy import or_,and_,not_ #AND条件 # emps = session.query(Employees).filter(and_(Employees.emp_no>10016,Employees.emp_no<10018)) # emps = session.query(Employees).filter(Employees.emp_no>10016).filter(Employees.emp_no<10018) # emps = session.query(Employees).filter((Employees.emp_no>10016) & (Employees.emp_no<10018)) # show(emps) #OR条件 # emps = session.query(Employees).filter(or_(Employees.emp_no>10018,Employees.emp_no<10003)) # emps = session.query(Employees).filter((Employees.emp_no>10018) | (Employees.emp_no<10003)) # show(emps) #Not # emps = session.query(Employees).filter(not_(Employees.emp_no<10018)) # emps = session.query(Employees).filter(~(Employees.emp_no<10018)) # show(emps) #注意:与或非的运算符&、、~ ,一定要在表达式上加上括号 #in # emplist = [10010,10005,10018] # emps = session.query(Employees).filter(Employees.emp_no.in_(emplist)) # show(emps) #not in # emplist = [10010,10005,10018] # emps = session.query(Employees).filter(~Employees.emp_no.in_(emplist)) # show(emps) #like,ilike可以忽略大小写 # emps = session.query(Employees).filter(Employees.first_name.like('t%')) # show(emps)
排序
#排序---升序、降序,默认升序aa emps = session.query(Employees).filter(Employees.emp_no>10010).order_by(Employees.emp_no) emps = session.query(Employees).filter(Employees.emp_no>10010).order_by(Employees.emp_no.asc()) emps = session.query(Employees).filter(Employees.emp_no>10010).order_by(Employees.emp_no.desc()) show(emps) #多列排序 #多列排序 emps = session.query(Employees).filter(Employees.emp_no>10010).order_by(Employees.last_name).order_by(Employees.emp_no.desc()) show(emps)
分页
#分页 emps = session.query(Employees).limit(4) show(emps) emps = session.query(Employees).limit(4).offset(2) show(emps)
消费者方法
#消费者方法 emps = session.query(Employees) print(len(list(emps))) #返回大量的结果集 print(emps.count()) #聚合函数count(*)的查询 #取所有数据 print(emps.all()) #返回列表,查不到返回空列表 #取首行 print(emps.first()) #返回首行,查不到返回None #有且只有一行,查到结果是多行会报错 print(emps.limit(1).one()) #删除 delete by query session.query(Employees).filter(Employees.emp_no>10018).delete() # session.commit() #提交则删除
聚合、分组
#聚合函数 from sqlalchemy import func #count query = session.query(func.count(Employees.emp_no)) print(query.one()) #只能有一个结果 print(query.scalar()) #取one()返回元祖的第一个元素 #max,min,avg print(session.query(func.max(Employees.emp_no)).scalar()) print(session.query(func.min(Employees.emp_no)).scalar()) print(session.query(func.avg(Employees.emp_no)).scalar()) #分组 print(session.query(Employees.gender,func.count(Employees.emp_no)).group_by(Employees.gender).all()) #[(<MyEnum.M: 'M'>, 12), (<MyEnum.F: 'F'>, 8)]
关联查询
#1 CREATE TABLE `departments` ( `dept_no` char(4) NOT NULL, `dept_name` varchar(40) NOT NULL, PRIMARY KEY (`dept_no`), UNIQUE KEY `dept_name` (`dept_name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; #2 CREATE TABLE `employees` ( `emp_no` int(11) NOT NULL, `birth_date` date NOT NULL, `first_name` varchar(14) NOT NULL, `last_name` varchar(16) NOT NULL, `gender` enum('M','F') NOT NULL, `hire_date` date NOT NULL, PRIMARY KEY (`emp_no`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; #3 CREATE TABLE `dept_emp` ( `emp_no` int(11) NOT NULL, `dept_no` char(4) NOT NULL, `from_date` date NOT NULL, `to_date` date NOT NULL, PRIMARY KEY (`emp_no`,`dept_no`), KEY `dept_no` (`dept_no`), CONSTRAINT `dept_emp_ibfk_1` FOREIGN KEY (`emp_no`) REFERENCES `employees` (`emp_no`) ON DELETE CASCADE, CONSTRAINT `dept_emp_ibfk_2` FOREIGN KEY (`dept_no`) REFERENCES `departments` (`dept_no`) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
from sqlalchemy import create_engine,Column,Integer,String,Date,Enum,ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import enum connstr = "{}://{}:{}@{}:{}/{}".format( "mysql+pymysql","wayne","wayne", "192.168.10.166",3306,"test" ) engine = create_engine(connstr,echo=True) Base = declarative_base(bind=engine) Session = sessionmaker() session = Session() class MyEnum(enum.Enum): M = 'M' F = 'F' class Employees(Base): __tablename__ = "employees" emp_no = Column(Integer,primary_key=True) birth_date = Column(Date,nullable=False) first_name = Column(String(14),nullable=False) last_name = Column(String(16)) gender = Column(Enum(MyEnum),nullable=False) hire_date = Column(Date,nullable=True) def __repr__(self): return "{} no={} name={} gender={}".format( self.__class__.__name__,self.emp_no,self.first_name,self.last_name, self.gender.value ) class Department(Base): __tablename__ = 'dapartments' dept_no = Column(String(4),primary_key=True) dept_name = Column(String(40),nullable=False,unique=True) def __repr__(self): return "{} no={} name={}".format( type(self).__name__,self.dept_no,self.dept_name ) class Dept_emp(Base): __tablename__ = "dept_emp" #ForeignKey定义外键约束 emp_no = Column(Integer,ForeignKey('empployes.emp_no',ondelete='CASCADE'),primary_key=True) dept_no = Column(String(4),ForeignKey('department.dept_no',ondelete='CASCADE'),primary_key=True) from_date = Column(Date,nullable=False) to_date = Column(Date,nullable=False) def __repr__(self): return "{} empno={} deptno={}".format(type(self).__name__,self.emp_no,self.dept_no)
1.使用隐式内连接
#查询10010员工所在的部门编号 results = session.query(Employees,Dept_emp).filter(Employees.emp_no == Dept_emp.emp_no).filter(Employees.emp_no == 10010).all() show(results) #运行结果 (Employees no=10010 name=Duangkaew gender=Piveteau, Dept_emp empno=10010 deptno=d004) (Employees no=10010 name=Duangkaew gender=Piveteau, Dept_emp empno=10010 deptno=d006)
SELECT * FROM employees, dept_emp WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s
2.使用join
#查询10010员工所在的部门编号 #1.隐式内连接 results = session.query(Employees,Dept_emp).filter(Employees.emp_no == Dept_emp.emp_no).filter(Employees.emp_no == 10010).all() show(results) #2join连接 results = session.query(Employees).join(Dept_emp,Employees.emp_no == Dept_emp.emp_no).filter(Employees.emp_no == 10010).all() print(results)
from sqlalchemy.orm import relationship class Employees(Base): __tablename__ = "employees" emp_no = Column(Integer,primary_key=True) birth_date = Column(Date,nullable=False) first_name = Column(String(14),nullable=False) last_name = Column(String(16)) gender = Column(Enum(Gender),nullable=False) hire_date = Column(Date,nullable=True) departments = relationship('Dept_emp') def __repr__(self): return "<Emp> no={} name={} gender={} {}".format( self.emp_no,self.first_name,self.last_name,self.departments )
#查询信息 # results = session.query(Employees).join(Dept_emp).filter(Employees.emp_no == Dept_emp.emp_no).filter(Employees.emp_no == 10010) # results = session.query(Employees).join(Dept_emp,Employees.emp_no == Dept_emp.emp_no).filter(Employees.emp_no == 10010) results = session.query(Employees).join(Dept_emp,(Employees.emp_no == Dept_emp.emp_no)&(Employees.emp_no == 10010)) print(results.all()) #结果如下: [<Emp> no=10010 name=Duangkaew gender=Piveteau [<Dept_emp> empno=10010 deptno=d004, <Dept_emp> empno=10010 deptno=d006]]
class Employees(Base): __tablename__ = "employees" emp_no = Column(Integer,primary_key=True) birth_date = Column(Date,nullable=False) first_name = Column(String(14),nullable=False) last_name = Column(String(16)) gender = Column(Enum(Gender),nullable=False) hire_date = Column(Date,nullable=True) departments = relationship('Dept_emp') def __repr__(self): return "<Emp> no={} name={} gender={} {}".format( self.emp_no,self.first_name,self.last_name,self.departments #修改self.departments为self.emp_no )
总结