Sqlalchemy ORM操作归类
#简单查询
#注意User是一个类对象,user_table是数据库中的表
#session = sessionmaker() #创建了一个自定义了的 Session类
1. session.query(User).all()
session.query(User).first()# 记录不存在时，first() 会返回 None
session.query(User).one()#用于获取所有元素，如果没有获得结果或者返回了多个结果，则会产生一个 error
session.query(User).count()#计数
session.query(User).scalar()#返回列表和标量
2. session.query(User).limit(2).all()#最多返回两条数据
session.query(User).offset(3).all()#从第4条数据开始
session.query(User).order_by(User.name).all()#排序
session.query(User).order_by(‘name‘).all()
session.query(User).order_by(User.name.desc()).all()
session.query(User).order_by(‘name desc‘).all()
3. filter/ filter_by过滤
equals:
session.query(User).filter(User.name == ‘ed‘)
not equals:
session.query(User).filter(User.name != ‘ed‘)
LIKE:
session.query(User).filter(User.name.like(‘%ed%‘))
IN:
session.query(User).filter(User.name.in_([‘ed‘, ‘wendy‘, ‘jack‘]))
# 可以和 query 对象协同工作：
session.query(User).filter(User.name.in_(session.query(User.name).filter(User.name.like(‘%ed%‘))))
NOT IN:
session.query(User).filter(~User.name.in_([‘ed‘, ‘wendy‘, ‘jack‘]))
IS NULL:
session.query(User).filter(User.name == None)
IS NOT NULL:
filter(User.name != None)
AND:
from sqlalchemy import and_
session.query(User).filter(and_(User.name == ‘ed‘, User.fullname ==‘Ed Jones‘))
# 或者连续调用 filter()/filter_by()两次
session.query(User).filter(User.name == ‘ed‘).filter(User.fullname== ‘Ed Jones‘)
OR:
from sqlalchemy import or_
session.query(User).filter(or_(User.name == ‘ed‘, User.name ==‘wendy‘))
match:
session.query(User).filter(User.name.match(‘wendy‘))
match 的参数内容由数据库后台指定。
4. 聚合查询
session.query(func.count(‘*‘)).select_from(User).scalar()
session.query(func.count(‘1‘)).select_from(User).scalar()
session.query(func.count(User.id)).scalar()
session.query(func.count(‘*‘)).filter(User.id> 0).scalar() # filter() 中包含 User，因此不需要指定表
session.query(func.count(‘*‘)).filter(User.name== ‘a‘).limit(1).scalar() == 1 # 可以用 limit() 限制 count() 的返回数
session.query(func.sum(User.id)).scalar()
session.query(func.now()).scalar()# func 后可以跟任意函数名，只要该数据库支持
session.query(func.current_timestamp()).scalar()
session.query(func.md5(User.name)).filter(User.id== 1).scalar()
5. 更多查询操作
#多条件查询
print(session.query(User).filter(and_(User.name.like("user),User.fullname.like("first%"))).all()
---相当于SQL语句--->
select * from user_table where user_table.name like %user anduser_table.fullname like first%
print(session.query(User).filter(or_(User.name.like("user%),User.password!= None)).all()
---相当于SQL语句--->
select * from user_table where user_table.name = user% oruser_table.password != none
#sql过滤
print(session(User).filter("id>:id").params(id=1).all()
---相当于SQL语句--->
select * from user_table where user_table.id > 1
#关联查询
print(session.query(User,Address).filter(User.id ==Address.user_id).all()
---相当于SQL语句--->
select * from user_table,address_table where user_table.id ==address.user_id
print (session.query(User).jion(User.address).all()
---相当于SQL语句--->
********************
print (session.query(User).outerjoin(User.address).all())
*******************
#子查询
stmt = session.query(Address.user_id,func.count(‘*‘).label("address_count").groupby(Address.user_id).subquery()
print(session.query(User,stmt.c.address_count).outjion((stmt,User.id ==stmt.c.user_id.order_by(User_id).all()
#exits
print (session.query(User).filter(exists().where(Address.user_id ==User.id)))
print (session.query(User).filter(User.addresses.any()))
其他《增删改》
query.filter(User.id == 1).update({User.name: ‘c‘})
user = query.get(1)
print user.name
user.name = ‘d‘
session.flush() # 写数据库，但并不提交
print query.get(1).name
session.delete(user)
session.flush()
print query.get(1)
session.rollback()
print query.get(1).name
query.filter(User.id == 1).delete()
session.commit()
print query.get(1)