flask,SQLAlchemy的in和not in等查询方式总结
官方参考：https://docs.sqlalchemy.org/en/13/orm/tutorial.html#common-filter-operators
in 查询， table.id.in_(list_ids)
[Shoe.query.filter_by(id=id).one() for id in my_list_of_ids]
shoes = Shoe.query.filter(Shoe.id.in_(my_list_of_ids)).all()
[next(s for s in shoes if s.id == id) for id in my_list_of_ids]
not in, 使用～符号，或者not_,(from sqlalchemy.sql.expression)：
~User.id.in_([123,456])
not_(User.id.in_([123,456]))
eg1.
id_list = [1, 2, 3, 4, 5] # in most case we have an integer list or set
s = text('SELECT id, content FROM myTable WHERE id IN :id_list')
conn = engine.connect() # get a mysql connection
rs = conn.execute(s, id_list=tuple(id_list)).fetchall()
eg2.
ids_list = [1,2,3]
query = "SELECT id, name FROM user WHERE id IN %s"
args = [(ids_list,)] # Don't forget the "comma", to force the tuple
conn.execute(query, args)
eg3.
user.id.notin_(tuple(list_ids)).update(data,synchronize_session=False)
eg4.
query = query.where(not_(user_model.User.email.like(email)))
eg5.
def get_all_batches(self):
query = sqlalchemy.select([
self._batches,
sqlalchemy.not_(sqlalchemy.exists(
sqlalchemy.select([1]).where(sqlalchemy.and_(
self._batches.c.id == self._batch_entries.c.batch_id,
self._batch_entries.c.crtsh_id == self._certs.c.crtsh_id,
self._certs.c.expiration_date > datetime.datetime.utcnow(),
self._certs.c.revoked_at.is_(None),
))
))
])
rows = self._engine.execute(query)
return [
Batch(
id=id,
description=description,
completed=completed,
)
for id, description, completed in rows
]
几种常见sqlalchemy查询：
#简单查询
print(session.query(User).all())
print(session.query(User.name, User.fullname).all())
print(session.query(User, User.name).all())
#带条件查询
print(session.query(User).filter_by(name='user1').all())
print(session.query(User).filter(User.name == "user").all())
print(session.query(User).filter(User.name.like("user%")).all())
#多条件查询
print(session.query(User).filter(and_(User.name.like("user%"), User.fullname.like("first%"))).all())
print(session.query(User).filter(or_(User.name.like("user%"), User.password != None)).all())
#sql过滤
print(session.query(User).filter("id>:id").params(id=1).all())
#关联查询
print(session.query(User, Address).filter(User.id == Address.user_id).all())
print(session.query(User).join(User.addresses).all())
print(session.query(User).outerjoin(User.addresses).all())
#聚合查询
print(session.query(User.name, func.count('*').label("user_count")).group_by(User.name).all())
print(session.query(User.name, func.sum(User.id).label("user_id_sum")).group_by(User.name).all())
#子查询
stmt = session.query(Address.user_id, func.count('*').label("address_count")).group_by(Address.user_id).subquery()
print(session.query(User, stmt.c.address_count).outerjoin((stmt, User.id == stmt.c.user_id)).order_by(User.id).all())
#exists
print(session.query(User).filter(exists().where(Address.user_id == User.id)))
print(session.query(User).filter(User.addresses.any()))
限制返回字段查询
person = session.query(Person.name, Person.created_at,
Person.updated_at).filter_by(name="zhongwei").order_by(
Person.created_at).first()
记录总数查询：
from sqlalchemy import func
# count User records, without
# using a subquery.
session.query(func.count(User.id))
# return count of user "id" grouped
# by "name"
session.query(func.count(User.id)).group_by(User.name)
from sqlalchemy import distinct
# count distinct "name" values
session.query(func.count(distinct(User.name)))