sqlachemy查询

 
 
语句
 
0.选择select
query
1.过滤(where)
filter:
空值和非空筛选is_  .isnot
dao_session.sub_session().query(XcEbikeGfence2.id).filter(XcEbikeGfence2.type == 1XcEbikeGfence2.deletedAt.is_(None)).all()
filter_by:
dao_session.session.time_db().query(TConfig).filter_by(service_id=service_id, root_router=router)
or筛选
filter(TAlarmTickets.tenant_id == tenant_id, TAlarmTickets.service_id == service_id, or_(TAlarmTickets.state == FixState.TO_FIX.value, TAlarmTickets.state == FixState.FIXING.value and TAlarmTickets.operate_pin == pin))
and筛选
dao_session.session.tenant_db().query(
TRidingCard
).filter(
TRidingCard.pin == pin,
TRidingCard.state == UserRidingCardState.USING.value,
TRidingCard.iz_total_times == 0,
or_(and_(TRidingCard.last_use_time < zeroToday,
TRidingCard.last_use_time > lastToday),
TRidingCard.last_use_time is None),
).update(
{
“remain_times”: TRidingCard.rece_times,
“last_use_time”: datetime.now()
}
((TFixTickets.state == 0) | ( (TFixTickets.state == FixState.FIXING.value) & (TFixTickets.operate_pin == pin))))
between范围
filter(TMoveOperation.created_at.between(start_time / 1000, end_time / 1000))
in
filter(TAlarmTickets.tenant_id == tenant_id, TAlarmTickets.car_id == car_id, TAlarmTickets.state.in_(FixState.unfixed_list()))
= is 值，.isnot
filter(XcEbikeGfence2.type == 1,XcEbikeGfence2.deletedAt.is_(None))
2.groupby分组
.filter( *car_filters).group_by(sql.func.date(XcMieba2CarAnalysis.created_at))
3.order排序
.order_by(TMoveOperation.end_time.desc()).limit(show_num)
4.having分组筛选
.group_by(User.age).having(User.age < 18)
5.contains
m = m.filter(TFixTickets.extra_info.contains(str(tp)))
 
5.添加筛选条件（多项）
        query_filter = set()        
       query_filter = query_filter | {TChangeBattery.tenant_id == tenant_id, TChangeBattery.service_id == service_id,                          TChangeBattery.open_bat_box_time.between(start_time, end_time)}
 
        query_filter = set()
        query_filter.add(TChangeBattery.tenant_id == tenant_id)
        query_filter.add(TChangeBattery.service_id.in_(service_ids))       
        if car_id:
            query_filter.add(TChangeBattery.car_id == car_id)
        res = dao_session.session.tenant_db().query(TChangeBattery).filter(*query_filter).order_by(
            TChangeBattery.created_at.desc()).all()
 
6.批量插入
 
 
 
# 常见函数
 
from sqlalchemy import func,
from sqlalchemy import sql, or_ and_
计数：func.count
当前时间:func.now()
聚合：func.GROUP_CONCAT,func.group_concat
求和:sql.func.sum
非空：sql.func.ifnull  给空值赋值  
sql.func.ifnull(sql.func.sum(XcEbikeUserOrder.cost) / 100, 0)
日期： sql.func.date
绝对值：sql.func.abs