flask sqlalchemy 操作

1 新增

user = User(mobile='15612345678', name='itcast')
db.session.add(user)
db.session.commit()
批量添加
db.session.add_all([user1, user2, user3])
db.session.commit()

db.session 是SQLAlchemy在每个事务中用来记录数据库操作的对象

2 查询

all()

#查询所有，返回列表
User.query.all()
db.session.query(User).all()

first()：

#查询第一个，返回对象  
User.query.first()
db.session.query(User).first()

get()

#根据主键ID获取对象，若主键不存在返回None
User.query.get(2)
db.session.query(User).get(2)

过滤

filter_by
User.query.filter_by(mobile='13911111111', id=1).first()  # and关系 只适用等于关系
filter
User.query.filter(User.mobile=='13911111111').first()

逻辑

from sqlalchemy import or_
User.query.filter(or_(User.mobile=='13911111111', User.name.endswith('号'))).all()
from sqlalchemy import and_
User.query.filter(and_(User.name != '13911111111', User.mobile.startswith('185'))).all()
from sqlalchemy import not_
User.query.filter(not_(User.mobile == '13911111111')).all()

offset limit

User.query.offset(2).all()
User.query.limit(3).all()

排序

User.query.order_by(User.id).all()  # 正序
User.query.order_by(User.id.desc()).all()  # 倒序

复合查询

User.query.filter(User.name.startswith('13')).order_by(User.id.desc()).offset(2).limit(5).all()

优化查询

# 查询数据库指明字段时，使用options(load_only(字段))
from sqlalchemy.orm import load_only
User.query.options(load_only(User.name, User.mobile)).filter_by(id=1).first() # 查询特定字段

聚合查询

from sqlalchemy import func
db.session.query(Relation.user_id, func.count(Relation.target_user_id)).filter(Relation.relation == Relation.RELATION.FOLLOW).group_by(Relation.user_id).all()

关联查询

• 第一种 ForeignKey

# db.ForeignKey('真实表名.user_id')，额外声明的字段属性，不对应数据库表的字段，不是映射字段，只是为了方便进行关联查询的时候使用
# 如果不做额外的声明，此种方式的查询属于惰性查询，user = User.query.get(1)  不查询UserProfile对应的表，直到要获取user.profile 属性时，才进行UserProfile的查询
class User(db.Model):
    ...
    profile = db.relationship('UserProfile', uselist=False)
    followings = db.relationship('Relation')

class UserProfile(db.Model):
    id = db.Column('user_id', db.Integer, db.ForeignKey('user_basic.user_id'), primary_key=True,  doc='用户ID')
    ...

class Relation(db.Model):
    user_id = db.Column(db.Integer, db.ForeignKey('user_basic.user_id'), doc='用户ID')
    ...

• 第二种 primaryjoin

class User(db.Model):
    ...

    profile = db.relationship('UserProfile', primaryjoin='User.id==foreign(UserProfile.id)', uselist=False)
    followings = db.relationship('Relation', primaryjoin='User.id==foreign(Relation.user_id)')

指定字段关联查询

# select a.user_id,a.target_user_id,b.user_name from user_relation as a inner join user_basic as b on a.target_user_id=b.user_id where a.user_id=1;
# 上面 sql 语句等同 下面orm语句
from sqlalchemy.orm import load_only, contains_eager
Relation.query.join(Relation.target_user).options(load_only(Relation.target_user_id), contains_eager(Relation.target_user).load_only(User.name)).all()

事务

# 在flask 上下文中，默认开启事务，需要使用commit提交事务，如果一个事务中，后条orm语句需要前面orm语句结果，可以使用db.session.flush(),将db.session记录的sql传到数据库中执行
environ = {'wsgi.version':(1,0), 'wsgi.input': '', 'REQUEST_METHOD': 'GET', 'PATH_INFO': '/', 'SERVER_NAME': 'itcast server', 'wsgi.url_scheme': 'http', 'SERVER_PORT': '80'}

with app.request_context(environ):
    try:
        user = User(mobile='18911111111', name='itheima')
        db.session.add(user)
        db.session.flush() # 将db.session记录的sql传到数据库中执行
        profile = UserProfile(id=user.id)
        db.session.add(profile)
        db.session.commit()
    except:
        db.session.rollback()