zoukankan      html  css  js  c++  java
  • 元编程

    元编程

     

    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
            )

    总结

     

    做一枚奔跑的老少年!
  • 相关阅读:
    Tcp连接和断开
    centos7服务器监控之nmon
    Jmeter之命令行生成HTML报告
    Centos7安装配置----1配置网络
    Wireshark抓包笔录--之指定IP地址筛选捕获结果
    关于如何刷新清除某个特定网站的缓存---基于Chrome浏览器
    关于如何清除某个特定网站的缓存---基于Chrome浏览器
    Linux 查看端口机服务
    XSS学习笔记
    安全测试学习笔记
  • 原文地址:https://www.cnblogs.com/xiaoshayu520ly/p/11196227.html
Copyright © 2011-2022 走看看