from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,ForeignKey,UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker,relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email?protected]:3306/db1",max_overflow=5)Base = declarative_base()# 创建单表class UserType(Base): __tablename__ = ‘usertype‘ nid = Column(Integer,primary_key=True,autoincrement=True) caption = Column(String(50),index=True)class Users(Base): __tablename__ = ‘users‘ id = Column(Integer,autoincrement=True) name = Column(String(32)) extra = Column(String(16)) usertype_id = Column(Integer,ForeignKey("usertype.nid")) __table_args__ = ( UniqueConstraint(‘id‘,‘name‘,name=‘uix_id_name‘),Index(‘ix_name‘,‘name‘),)#创建表格 删除表格def init_db(): Base.metadata.create_all(engine)def drop_db(): Base.metadata.drop_all(engine)# init_db()Session = sessionmaker(bind=engine)session = Session()#####################增####################### obj = UserType(name="alex0")# # session.add(obj)# session.add_all([# # UserType(caption="黑金"),# # UserType(caption="黑钻"),# UserType(caption="普通"),# UserType(caption="白金"),# ])#####################删####################### session.query(UserType).filter(UserType.nid > 2).delete()#####################改####################### session.query(UserType).filter(UserType.nid < 2).update({"caption" : "099"})# session.query(UserType).filter(UserType.nid == 2).update({UserType.caption: UserType.caption + "099"},synchronize_session=False)# session.query(UserType).filter(UserType.nid == 2).update({"num": Users.num + 1},synchronize_session="evaluate")#####################改####################### ret = session.query(UserType).all()# # for i in ret:# # print(i.caption,i.nid)# # ret = session.query(Users.name,Users.extra).all()# # ret = session.query(Users).filter_by(name=‘11‘).all()# # ret = session.query(Users).filter_by(name=‘11‘).first()# session.commit()# session.close()# 条件# ret = session.query(Users).filter_by(name=‘alex‘).all()# ret = session.query(Users).filter(Users.id > 1,Users.name == ‘eric‘).all()# ret = session.query(Users).filter(Users.id.between(1,3),Users.name == ‘eric‘).all()# ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()# ret = session.query(Users).filter(~Users.id.in_([1,4])).all()# ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=‘eric‘))).all()# from sqlalchemy import and_,or_# ret = session.query(Users).filter(and_(Users.id > 3,Users.name == ‘eric‘)).all()# ret = session.query(Users).filter(or_(Users.id < 2,Users.name == ‘eric‘)).all()# ret = session.query(Users).filter(# or_(# Users.id < 2,# and_(Users.name == ‘eric‘,Users.id > 3),# Users.extra != ""# )).all()### # 通配符# ret = session.query(Users).filter(Users.name.like(‘e%‘)).all()# ret = session.query(Users).filter(~Users.name.like(‘e%‘)).all()## # 限制# ret = session.query(Users)[1:2]## # 排序# ret = session.query(Users).order_by(Users.name.desc()).all()# ret = session.query(Users).order_by(Users.name.desc(),Users.id.asc()).all()## # 分组# from sqlalchemy.sql import func## ret = session.query(Users).group_by(Users.extra).all()# ret = session.query(# func.max(Users.id),# func.sum(Users.id),# func.min(Users.id)).group_by(Users.name).all()## ret = session.query(# func.max(Users.id),# func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()## # 连表## ret = session.query(Users,Favor).filter(Users.id == Favor.nid).all()## ret = session.query(Person).join(Favor).all()#连表 join on# ret = session.query(Person).join(Favor,isouter=True).all()### # 组合# q1 = session.query(Users.name).filter(Users.id > 2)# q2 = session.query(Favor.caption).filter(Favor.nid < 2)# ret = q1.union(q2).all()## q1 = session.query(Users.name).filter(Users.id > 2)# q2 = session.query(Favor.caption).filter(Favor.nid < 2)# ret = q1.union_all(q2).all()
