标签：笔记 sqlalchemy
查询全部记录，返回列表
user = self.dbSession.query(User).all() #返回列表，全部结果封装在列表中
user = self.dbSession.query(User.username, User.createtime).all() #返回指定字段，每条记录封装在元组中
user = self.dbSession.query(House).all() #在添加到add后没有commit前的数据都可以查出来
user = self.dbSession.query(House, User).all() #可以把两个表的数据都查询出来
查询一条记录，返回的是一个对象
user = self.dbSession.query(User).filter(User.password == ‘222‘).first() #返回的是一个对象，不存在返回None
user = self.dbSession.query(User).filter(User.password == ‘444‘).one() #返回的是一个对象，如果返回多条会报错
user = self.dbSession.query(User).filter(User.password == ‘333‘).one() #返回一条不会报错
user = self.dbSession.query(User).filter(text(‘password="222"‘)).all()
user = self.dbSession.query(User).get(2) #通过主键获取数据
查询记录返回条数
user = self.dbSession.query(User).filter(text(‘password="222"‘)).limit(2).all() #返回两条记录
user = self.dbSession.query(User).filter(text(‘password="222"‘)).offset(2).all()#从第三条记录开始返回
user = self.dbSession.query(User).filter(text(‘password="222"‘)).slice(1,3).all() #切片返回记录
返回第一个结果的第一个元素
user = self.dbSession.query(func.count(‘*‘)).select_from(User).scalar()
使用sql语句进行条件查询注：需要用类名指定字段名
user = self.dbSession.query(User).filter(User.username.like("zhan%")).all()
user = self.dbSession.query(User).filter(User.username == "zhansan").all()
user = self.dbSession.query(User).filter(User.password == ‘222‘, User.username == ‘c‘).all() #写入两个条件
user = self.dbSession.query(User).filter(User.password == ‘222‘).filter(User.username=="c").all()
进行and_, or_ 查询
from sqlalchemy import and_, or_
user = self.dbSession.query(User).filter(and_(User.username.like("zha%"),
User.password.like("22%"))).all()
user = self.dbSession.query(User).filter(or_(User.username.like("zha%"),
User.password.like("33%"))).all()
user = self.dbSession.query(User).filter(or_(User.username.like("zha%"),
User.password != ‘333‘)).all()
user = self.dbSession.query(User).filter(or_(User.username.like("zha%"),
User.createtime < datetime.now())).all()
进行in_ 查询
user = self.dbSession.query(User).filter(User.id.in_((3, 2))).all() #查询在in中的数据
user = self.dbSession.query(User).filter(User.id.notin_((3, 2))).all() #查询不在notin中的数据
进行not查询
user = self.dbSession.query(User).filter(not_(User.last_login==None)) #获取不是null的值
user = self.dbSession.query(User).filter(User.last_login != None).all() #
使用sql语句进行过滤查询
user = self.dbSession.query(User.username).filter(text("id > :id")).params(id=2) #文本表达式需要写text()
user = self.dbSession.query(User.username).filter(User.id > 2).all() #与上面结果相同
order_by查询
user = self.dbSession.query(User).filter(User.password==‘222‘).order_by(-User.id).all()
user = self.dbSession.query(User).filter(User.password==‘222‘).order_by(User.id.desc()).all()
user = self.dbSession.query(User).filter(User.password == ‘222‘).order_by(text(‘id desc‘)).all()
user = self.dbSession.query(User).order_by(-User.id).all()
func使用
user = self.dbSession.query(func.count(‘*‘)).filter(User.id > 1) #id大于1的有多少条记录
user = self.dbSession.query(func.count(‘*‘)).select_from(User).all() #表中有多少条记录
user = self.dbSession.query(func.now()).scalar() #返回时间
分组查询 sum求总和，count求总行，min求最小，max求最大，avg求平均
func 的函数名，数据库支持就可以
label(‘aa‘)取别名
user = self.dbSession.query(User.loginnum,
func.count(User.id).label("a")).group_by(User.loginnum).all()
user = self.dbSession.query(User.loginnum,
func.sum(User.id).label("a")).group_by(User.loginnum).all()
user = self.dbSession.query(User.loginnum,
func.min(User.id).label("a")).group_by(User.loginnum).all()
user = self.dbSession.query(User.loginnum,
func.count(User.id).label("a")).group_by(User.loginnum).order_by(-User.loginnum).all()
按分钟分组
user = self.dbSession.query(extract(‘minute‘, User.createtime).label(‘minute‘),
func.count(‘*‘).label(‘count‘)).group_by(‘minute‘).all()
按天分组
user = self.dbSession.query(extract(‘day‘, User.last_login).label(‘day‘),
func.count(‘*‘).label(‘count‘)).group_by(‘day‘).all()
聚合查询having
user = self.dbSession.query(User.loginnum,
func.count(User.id)).group_by(User.loginnum).having(func.count(User.id) == 1).all()
filter_by() 使用关键字进行条件查询，不需要用类名指定字段名
user = self.dbSession.query(User).filter_by(username=‘zhansan‘).all()
user = self.dbSession.query(User).filter_by(password=‘222‘).all()
update () 更新操作，相当于执行了add操作
self.dbSession.query(User).filter(User.id==1).update({User.username:"aaa"})
self.dbSession.commit()
如果查询出一条记录后，把username修改一下，再进行查询时获取的是缓存区中的内容
aa = self.dbSession.query(User).filter(User.id==1).first()
aa.username = ‘bbb‘
print aa, aa.username
user1 = self.dbSession.query(User).all()
print user1, user1[0].username
delete后需要commit提交，在commit提交前再获取用户已经获取不到了，如果rollback一下后 又可以获取了
aa = self.dbSession.query(User).filter(User.id==3).first()
aa.username = ‘ddd‘
self.dbSession.delete(aa)
self.dbSession.rollback()
print aa, aa.username
user1 = self.dbSession.query(User).all()
print user1, user1[0].username
插入多条记录
self.dbSession.execute(User.__table__.insert(), [
{‘name‘: randint(1, 100), ‘username‘:‘aaa‘,‘loginnum‘: randint(1, 100)} for i in xrange(500)
])
self.dbSession.commit()
获取字段长度
print User.username.property.columns[0].type.length
子查询
第一种where子查询, 把内层查询结果当作外层查询的比较条件
subq = self.dbSession.query(Classes.id).filter(Classes.id == 3).subquery()
aa = self.dbSession.query(Student.username).filter(subq == Student.class_id )
print aa
打印的sql语句
SELECT student.username AS student_username
FROM student
WHERE student.class_id = (SELECT classes.id
FROM classes
WHERE classes.id = %(id_1)s)
第二种from子查询，把内层查询结果供外层再次查询，子查询要有别名
查询id为3 的班级的所有学生
subq = self.dbSession.query(Classes).filter(Classes.id == 3).subquery()
aa = self.dbSession.query(subq.c.id,Student.username).filter(subq.c.id == Student.class_id ).all()
print aa
打印的sql语句
SELECT anon_1.id AS anon_1_id, anon_1.classname AS anon_1_classname, anon_1.createtime AS anon_1_createtime
FROM (SELECT classes.id AS id, classes.classname AS classname, classes.createtime AS createtime
FROM classes
WHERE classes.id = %(id_1)s) AS anon_1, student
WHERE anon_1.id = student.class_id
第三种exists子查询， 外层查询的结果通过内层查询结果判断是否输出，内层查询有结果即为真True则外层查询输出结果，为假False则不输出
查询没有学生的班级
aa = self.dbSession.query(Classes).filter(exists().where(Classes.id==Student.class_id)).all()
print aa
打印的sql语句
SELECT classes.id AS classes_id, classes.classname AS classes_classname, classes.createtime AS classes_createtime
FROM classes
WHERE EXISTS (SELECT * FROM student WHERE classes.id = student.class_id)
查询没有学生的班级
第二种写法
subq = self.dbSession.query(Student).filter(Classes.id == Student.class_id)
aa = self.dbSession.query(Classes).filter(subq.exists()).filter().all()
print aa
打印的sql语句
SELECT classes.id AS classes_id, classes.classname AS classes_classname, classes.createtime AS classes_createtime
FROM classes
WHERE EXISTS (SELECT 1 FROM student WHERE classes.id = student.class_id)
表的5种约束
主键约束 primary key: 不允许为空，不允许重复
主键自动增长 auto_increment
唯一约束 unique
非空约束 not null
外键约束 foreign key
普通链接查询一对多
新建两张表
class Student(Base):
__tablename__ = ‘student‘
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
class_id = Column(Integer, ForeignKey(‘classes.id‘))
class Classes(Base):
__tablename__ = ‘classes‘
id = Column(Integer, primary_key=True, autoincrement=True)
classname = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
使用execute插入班级表数据
self.dbSession.execute(Classes.__table__.insert(), [
{‘classname‘: str(i) + ‘class‘} for i in xrange(1,4)
])
self.dbSession.commit()
使用execute插入学生表数据
self.dbSession.execute(Student.__table__.insert(), [
{‘username‘: ‘username‘ + str(i),
‘class_id‘: randint(1, 3)} for i in xrange(20)
])
self.dbSession.commit()
查询ID为1的班级的所有学生
aa = self.dbSession.query(Student, Classes).filter(Student.class_id==Classes.id, Student.class_id==1)
for student,classes in aa:
print student.username,student.class_id,classes.id
内链接查询，查询结果没有空字段
aa = self.dbSession.query(Student).join(Classes)
左链接查询，右链接就是换个方向，查询结果含有空字段
aa = self.dbSession.query(Student).outerjoin(Classes)
aa = self.dbSession.query(Student).join(Classes,isouter=True) #第二种方法
获取1班的学生，左连接查询时，想保留哪个表的数据就用哪个表作为左表
aa = self.dbSession.query(Student).outerjoin(Classes).filter(Classes.id==1).all()
获取2班的并且学号大于10的学生
aa = self.dbSession.query(Student).outerjoin(Classes).filter(Classes.id==2,Student.id>10).all()
获取2班的并且学号大于10的学生姓名和班级名
aa = self.dbSession.query(Student.username,Classes.classname).outerjoin(Classes).filter(Classes.id == 2, Student.id > 10).all()
查看学号为1的学生是哪个班的
aa = self.dbSession.query(Classes.id).outerjoin(Student).filter(Student.id == 1).all()
查看学号为1的学生是哪个班的，什么时间成立的班级
aa = self.dbSession.query(Classes.id,
Classes.createtime,
Student.username).outerjoin(Student).filter(Student.id == 1).all()
关系表在添加和删除数据的注意事项：
1向子表中添加数据时，如果子表外键字段设置非空约束，需要在添加数据时添加外键字段，否则无法插入数据
2父表在删除或更新时需要注意被关联外键的字段删除更新行为
3如果行为是限制删除那么就要先查询到子表然后删除子表数据再删除父表数据
在项目代码中插入数据
插入一个学生数据，需要有一个班级对象
classes = self.dbSession.query(Classes).filter(Classes.id==3).first()
student1 = Student(username=‘liubei‘, class_id= classes.id)
self.dbSession.add(student1)
self.dbSession.commit()
使用add_all批量添加数据
classes = self.dbSession.query(Classes).filter(Classes.id==3).first()
student1 = Student(username=‘liubei‘, class_id= classes.id)
student2 = Student(username=‘guanyu‘, class_id= classes.id)
student3 = Student(username=‘zhangfei‘, class_id= classes.id)
self.dbSession.add(student1)
self.dbSession.add_all([student1, student2, student3])
self.dbSession.commit()
建议使用的链接查询一对多
正向查询：
从子表查父表数据，具有ForeignKey字段的表是子表
反向查询：
从父表查子表数据
#经典联表查询
class Student(Base):
__tablename__ = ‘student‘
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
class_id = Column(Integer, ForeignKey(‘classes.id‘))
#反向查找需要加这句，backref=‘students‘,相当于在classes表里多了个字段students
classes = relationship("Classes", backref=‘students‘)
class Classes(Base):
__tablename__ = ‘classes‘
id = Column(Integer, primary_key=True, autoincrement=True)
classname = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
正向查询：先获取子表的查询结果，注：有ForeignKey的是子表 每条子表记录对应一条父表记录
students = self.dbSession.query(Student).filter(Student.id>10).all()
for student in students:
print student.classes.classname, student.classes.createtime
反向查询：先获取父表的查询结果，注：父表查询出来的每条记录都对应着多条子表记录
classes = self.dbSession.query(Classes).all()
classes=classes[0]
for student in classes.students:
print student.username, student.id, student.createtime
正向添加：
（在子表中赋值relationship的字段添加父表对象）
在项目代码中修改数据：
更新id为5号的学生的班级为2班，
1获取2班级对象
2获取5号学生对象
3给5号学生的班级字段赋值为2班级对象
4提交
classes = self.dbSession.query(Classes).filter(Classes.id == 2).first()
student = self.dbSession.query(Student).filter(Student.id==5).first()
student.classes = classes
self.dbSession.add(student)
self.dbSession.commit()
增加一个叫项羽的学生，班级设为在1班（）
（在子表中赋值relationship的字段添加父表对象）
1获取1班的对象
2新建一个学生对象
3新学生对象的名称赋值为项羽，班级字段赋值为1班的对象
4提交
classes = self.dbSession.query(Classes).filter(Classes.id == 1).first()
student = Student()
student.username = "项羽"
student.classes = classes
self.dbSession.add(student)
self.dbSession.commit()
反向添加：
（在父表的backref指定的字段添加子表数据）
1获取3班的对象
2新建一个学生对象
3新学生对象的名称赋值为项羽
4使用3班对象的外键关联字段添加新学生对象
4提交
classes = self.dbSession.query(Classes).filter(Classes.id == 3).first()
student = Student()
student.username = "赵云"
classes.students.append(student)
self.dbSession.add(classes)
self.dbSession.commit()
反向添加：
（在父表的backref指定的字段添加子表数据）
添加多个学生到2班
classes = self.dbSession.query(Classes).filter(Classes.id == 2).first()
student1 = Student(username="关羽")
student2 = Student(username="黄忠")
student3 = Student(username="魏延")
classes.students.extend([student1,student2,student3])
self.dbSession.add(classes)
self.dbSession.commit()
多对多查询
#多对多关系表
class StudentToScore(Base):
__tablename__ = ‘student_to_score‘
student_id = Column(Integer, ForeignKey(‘student.id‘), primary_key=True)
score_id = Column(Integer, ForeignKey(‘score.id‘), primary_key=True)
class Student(Base):
__tablename__ = ‘student‘
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
class_id = Column(Integer, ForeignKey(‘classes.id‘))
classes = relationship("Classes", backref=‘students‘)
class Score(Base):
__tablename__ = ‘score‘
id = Column(Integer, primary_key=True, autoincrement=True)
coursename = Column(String(50), nullable=False)
score = Column(Integer)
class Classes(Base):
__tablename__ = ‘classes‘
id = Column(Integer, primary_key=True, autoincrement=True)
classname = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
插入数据
self.dbSession.execute(Classes.__table__.insert(), [
{‘classname‘: str(i) + ‘班‘} for i in xrange(1,4)
])
self.dbSession.commit()
self.dbSession.execute(Student.__table__.insert(), [
{‘username‘: ‘username‘ + str(i+1),
‘class_id‘: randint(1, 3)} for i in xrange(20)
])
self.dbSession.commit()
self.dbSession.execute(Score.__table__.insert(), [
{‘coursename‘: ‘coursename‘+ str(i+1),
‘score‘: randint(60, 80)} for i in xrange(5)
])
self.dbSession.commit()
# self.dbSession.execute(StudentToScore.__table__.insert(), [
# {‘student_id‘: randint(1, 20),
# ‘score_id‘: randint(1, 5)} for i in xrange(10)
# ])
#self.dbSession.commit()
查询数据：
查询1号学员所有科目的分数
aa =self.dbSession.query(StudentToScore.score_id).filter(StudentToScore.student_id==1).all()
a = zip(*aa) #把[(1,), (2,)] 变成[(1, 2)]
print a
print self.dbSession.query(Score.score).filter(Score.id.in_(a[0])).all()
建议使用的链接查询多对多
创建表
class StudentToScore(Base):
__tablename__ = ‘student_to_score‘
student_id = Column(Integer, ForeignKey(‘student.id‘), primary_key=True)
score_id = Column(Integer, ForeignKey(‘score.id‘), primary_key=True)
class Student(Base):
__tablename__ = ‘student‘
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
class_id = Column(Integer, ForeignKey(‘classes.id‘))
classes = relationship("Classes", backref=‘students‘)
#新增句子
students_scores = relationship("Score", secondary=StudentToScore.__table__)
class Score(Base):
__tablename__ = ‘score‘
id = Column(Integer, primary_key=True, autoincrement=True)
coursename = Column(String(50), nullable=False)
score = Column(Integer)
#新增句子
scores_students = relationship("Student", secondary=StudentToScore.__table__)
class Classes(Base):
__tablename__ = ‘classes‘
id = Column(Integer, primary_key=True, autoincrement=True)
classname = Column(String(50), nullable=False)
createtime = Column(DateTime, default=datetime.now)
查询数据：
查询1号学员所有科目的分数
aa = self.dbSession.query(Student).filter(Student.id == 1).first()
print aa.students_scores
安装加密模块
pip install pbkdf2
本文出自 “骑猪去旅行” 博客，请务必保留此出处http://songqinglong.blog.51cto.com/7591177/1956847
标签：笔记 sqlalchemy
原文地址：http://songqinglong.blog.51cto.com/7591177/1956847
