1.sqlalchemy是一个orm框架,并且大量使用元编程
import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,Date,String connect_string = "{}://{}:{}@{}:{}/{}".format( 'mysql+pymysql', 'test', '1qazxsw2', '127.0.0.1', '3306', 'blog' ) engine = create_engine(connect_string,echo=True) Base = declarative_base() #创建基类便于实体类继承 class Student(Base): #指定表名 __tablename__ = 'student' #定义属性对应字段 id = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(64),nullable=False) age = Column(Integer) def __repr__(self): return '<{} id={},name={},age={}> '.format(self.__class__.__name__,self.id,self.name,self.age) #创建继承自base的所有表,此处创建表 Base.metadata.create_all(engine)
import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,Date,String from sqlalchemy.orm import sessionmaker #pymysql的链接引擎 connect_string = "{}://{}:{}@{}:{}/{}".format( 'mysql+pymysql', 'test', '1qazxsw2', '67.216.218.7', '3306', 'blog' ) engine = create_engine(connect_string,echo=True) Base = declarative_base() #创建基类便于实体类继承 class Student(Base): __tablename__ = 'student' id = Column(Integer,primary_key=True,autoincrement=True) name = Column(String(64),nullable=False) age = Column(Integer) def __repr__(self): return '<{} id={},name={},age={}> '.format(self.__class__.__name__,self.id,self.name,self.age) ''' 创建,删除继承自SQLAlchemy Base.metadata.create_all(engine) Base.metadata.drop_all(engine) ''' #在一个会话中操作数据库,会话建立在连接上,连接被引擎管理 #创建session Session = sessionmaker(bind=engine) #返回类 session = Session() #实例化 s1 = Student() #Base源码里有setattr反射方法 s1.id = 1 s1.name = 'harden' s1.age = 29 s2 = Student(name='curry') #源码里有__new__(*args,**kwargs) s2.id = 2 s2.age = 30 try: session.add(s1) session.commit() except: session.rollback() finally: session.close() #查询 ''' query将实体类传入,返回类的可迭代对象,这时候并不查询,迭代他就执行sql来查数据库 封装数据到指定类的实例 get方法使用主键查询,返回一条传入类的一个实例 ''' students = session.query(Student) #无条件查询 for student in students: print(student) print("~~~~~~~~~~~~~~~~~~~~~~~") query_stu = session.query(Student).get(1) #通过主键查询 print(query_stu) #更改,,,(先查-->修改后--->提交) update_student = session.query(Student).get(1) update_student.name = 'harden' session.add(update_student) session.commit() ''' 状态:每一个实体,都一个状态属性_sa_instance_state,其类型是sqlalchemy.orm.state.Instancestate,可以使用sqlalchemy。inspect(entity)查询状态 常见的状态值有:transient,pending,persistent,deleted,detached, transient:实体类尚未加入到session中,同时也没有保存到数据库中 pending:transient的实体被add()到session中,状态切换到pending中,但还没有flush到数据库中 persitent:session中的实体对象对应着数据库中的真是记录,pengding状态提交后可以变成persistent,或者查询成功返回的实体也是persistent deleted:实体被删除且已经flush但未commit完成,事务提交成功了,实体变成detached,失败变成persistent状态 detached:成功删除实体进入这个状态 新建一个实体,状态是transient临时的 一旦add()后从transient变成pending 成功commit()后从pendig变成persistent 成功查询返回的实体对象,也是persistent状态 persistent状态的实体,修改后依然是persistent persistent状态的实体,删除后,flush后但没有commit,就变成deleted状态,成功提交变成detached,提交失败返回persistent,flush主动把改变应用到数据库 删除,修改,需要对应一个真实的记录,所有要求实体对象是persistent状态 ''' def getstate(entity,i): insp = sqlalchemy.inspect(entity) state = "sessionid={},attached={} transient={},persisteng={} pending={},deleted={},detached={}".format( insp.session_id, insp._attached, insp.transient, insp.persistent, insp.pending, insp.deleted, insp.detached ) print(i,state) print(insp.key) ins_student = session.query(Student).get(1) getstate(ins_student)
import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column,Integer,Date,String,Enum from sqlalchemy.orm import sessionmaker import enum #pymysql的链接引擎 connect_string = "{}://{}:{}@{}:{}/{}".format( 'mysql+pymysql', 'test', '1qazxsw2', '67.216.218.7', '3306', 'test' ) class MyEnum(enum.Enum): M= 'M' F= 'F' engine = create_engine(connect_string,echo=True) Base = declarative_base() #创建基类便于实体类继承 class Employee(Base): __tablename__ = 'employees' emp_no = Column(Integer,primary_key=True) birth_date = Column(Date,nullable=False) first_name = Column(String(14),nullable=False) l_name = Column('last_name',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.gender.value) #创建session Session = sessionmaker(bind=engine) #返回类 session = Session() #实例化 def show(emps): for i in emps: print(i) print("~~~~~~~~~~~~~~~~show~~~~~~~~~~~~~~~~") # emps = session.query(Employee).filter(Employee.emp_no>10010).filter(Employee.gender==MyEnum.M) # show(emps) # emps = session.query(Employee).filter((Employee.emp_no>10010) & (Employee.gender == MyEnum.M)) # show(emps) # emps1 = session.query(Employee).filter(Employee.emp_no.in_([10010,10010,10012])) # emps2 = session.query(Employee).filter(Employee.l_name.like('P%')) # show(emps1) # show(emps2) # emps = session.query(Employee).filter(Employee.emp_no>10010).order_by(Employee.emp_no.desc()) # show(emps) # emps = session.query(Employee).limit(4) # show(emps) # emps = session.query(Employee) # print(emps.all()) #取回所有数据,返回列表,查不到返回空列表 # print(emps.first()) #首行,查不到返回None # print(emps.count()) #聚合函数count(*)的查询 #delete by query # session.query(Employee).filter(Employee.emp_no>10033).delete() #session.commit() 提交删除 from sqlalchemy import func # query = session.query(func.count(Employee.emp_no)) # print(query.one()) #只能有一行结构 # print(query.scalar()) #取one()返回元祖的第一个元素 print(session.query(func.max(Employee.emp_no)).scalar()) print(session.query(Employee.gender,func.count(Employee.emp_no)).group_by(Employee.gender).all())
group by
SELECT emp_no,SUM(salary),AVG(salary),COUNT(emp_no) FROM salaries WHERE emp_no<10003 GROUP BY emp_no 根据emp_no分条目显示 SELECT emp_no,SUM(salary),AVG(salary),COUNT(emp_no) FROM salaries 所有的都显示为一行